hephaestus/crates/heph-core/src/sqlite/tasks.rs
Erich Blume c81d45a291
Some checks failed
Build / validate (pull_request) Failing after 4s
heph-core: real HLC + persistent device origin (sync 8a)
First sync slice. Ratifies yrs for body merge in the tech-spec.

- hlc module: Hlc (physical, counter, origin) with a fixed-width encoding
  whose lexical order equals causal order; HlcClock generator (tick/update)
  — clock-injected, strictly monotonic. Unit + 2 proptests.
- meta table (migration v3) holds the stable per-device `origin` and the
  last HLC. next_hlc() does a read-modify-write inside the caller's
  transaction (store is single-writer, so no race), replacing the
  timestamp placeholder. Every node write is now stamped with a real,
  monotonic, causally-ordered HLC.

4 stamping integration tests (monotonic under stalled/regressed clock;
origin shared + persists across reopen; HLC resumes from persisted state).
89 tests green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-05-31 21:13:55 -07:00

364 lines
12 KiB
Rust

//! `tasks` table operations.
//!
//! A committed task is a `task` node plus a `tasks` row. On creation it also
//! gets a canonical context `doc` and a `canonical-context` link (tech-spec §6).
use rusqlite::{Connection, OptionalExtension, Row};
use super::{links, log, next_hlc, nodes};
use crate::error::{Error, Result};
use crate::model::{Attention, Health, LinkType, NewTask, NodeKind, Task, TaskState};
use crate::ranking::{self, RankedTask};
use crate::recurrence;
fn from_row(row: &Row) -> rusqlite::Result<Task> {
let attention = match row.get::<_, Option<String>>("attention")? {
Some(s) => Some(
Attention::parse(&s)
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?,
),
None => None,
};
Ok(Task {
node_id: row.get("node_id")?,
attention,
do_date: row.get("do_date")?,
late_on: row.get("late_on")?,
state: TaskState::parse(&row.get::<_, String>("state")?)
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?,
recurrence: row.get("recurrence")?,
})
}
const COLUMNS: &str = "node_id, attention, do_date, late_on, state, recurrence";
/// Create a committed task: its task node, the `tasks` row, the canonical
/// context doc, the `canonical-context` link, and (if given) an `in-project`
/// link — all in one transaction.
pub(super) fn create(conn: &mut Connection, owner: &str, now: i64, input: NewTask) -> Result<Task> {
let task = Task {
node_id: String::new(), // filled below
attention: input.attention,
do_date: input.do_date,
late_on: input.late_on,
state: TaskState::Outstanding,
recurrence: input.recurrence,
};
let tx = conn.transaction()?;
let task_hlc = next_hlc(&tx, now)?;
let task_node = nodes::build(
owner,
now,
&task_hlc,
NodeKind::Task,
input.title.clone(),
None,
);
nodes::insert(&tx, &task_node)?;
tx.execute(
"INSERT INTO tasks (node_id, attention, do_date, late_on, state, recurrence)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
(
&task_node.id,
task.attention.map(|a| a.as_str()),
task.do_date,
task.late_on,
task.state.as_str(),
&task.recurrence,
),
)?;
// The canonical context doc (the task's jumping-off point / checklist body).
let doc_hlc = next_hlc(&tx, now)?;
let doc = nodes::build(
owner,
now,
&doc_hlc,
NodeKind::Doc,
input.title.clone(),
Some(String::new()),
);
nodes::insert(&tx, &doc)?;
links::add(&tx, now, &task_node.id, &doc.id, LinkType::CanonicalContext)?;
if let Some(project_id) = &input.project_id {
links::add(&tx, now, &task_node.id, project_id, LinkType::InProject)?;
}
tx.commit()?;
Ok(Task {
node_id: task_node.id,
..task
})
}
/// Fetch a task by node id.
pub(super) fn get(conn: &Connection, node_id: &str) -> Result<Option<Task>> {
let task = conn
.query_row(
&format!("SELECT {COLUMNS} FROM tasks WHERE node_id = ?1"),
[node_id],
from_row,
)
.optional()?;
Ok(task)
}
fn require(conn: &Connection, node_id: &str) -> Result<Task> {
get(conn, node_id)?.ok_or_else(|| Error::NodeNotFound(node_id.to_string()))
}
/// Set a task's lifecycle state. Completing a **recurring** task rolls it
/// forward in place (tech-spec §4.4) rather than marking it done.
pub(super) fn set_state(
conn: &mut Connection,
owner: &str,
now: i64,
node_id: &str,
state: TaskState,
) -> Result<Task> {
let task = require(conn, node_id)?;
if state == TaskState::Done && task.recurrence.is_some() {
return roll_forward(conn, owner, now, &task);
}
conn.execute(
"UPDATE tasks SET state = ?1 WHERE node_id = ?2",
(state.as_str(), node_id),
)?;
nodes::touch(conn, now, node_id)?;
require(conn, node_id)
}
/// Roll a recurring task forward on completion (tech-spec §4.4): reset its
/// checklist to all-unchecked, log the occurrence, and advance the do-date to
/// the next RRULE instance strictly after `now` (skipping misses) — all in one
/// transaction. If the series is exhausted, the task is finally marked done.
fn roll_forward(conn: &mut Connection, owner: &str, now: i64, task: &Task) -> Result<Task> {
let rrule = task
.recurrence
.as_deref()
.expect("roll_forward called on a recurring task");
let tx = conn.transaction()?;
// 1. Fresh checklist — reset the canonical context doc's checkboxes.
if let Some(doc_id) = links::first_dst(&tx, &task.node_id, LinkType::CanonicalContext)? {
if let Some(doc) = nodes::get(&tx, &doc_id)? {
let body = doc.body.unwrap_or_default();
let reset = recurrence::reset_checkboxes(&body);
if reset != body {
let hlc = next_hlc(&tx, now)?;
tx.execute(
"UPDATE nodes SET body = ?1, modified_at = ?2, hlc = ?3 WHERE id = ?4",
(&reset, now, hlc, &doc_id),
)?;
links::sync_wiki_links(&tx, owner, &doc_id, &reset, now)?;
}
}
}
// 2. Narrative history — append the completed occurrence to the log.
let entry = match task.do_date {
Some(d) => format!("- completed occurrence (do-date {d})"),
None => "- completed occurrence".to_string(),
};
log::append(&tx, owner, now, &task.node_id, &entry)?;
// 3. Advance the do-date (or finally finish a finite series).
advance(&tx, now, &task.node_id, rrule, task.do_date)?;
nodes::touch(&tx, now, &task.node_id)?;
tx.commit()?;
require(conn, &task.node_id)
}
/// Advance a recurring task to its next instance after `now`, or mark it `done`
/// if the series is exhausted. Shared by completion roll-forward and `skip`.
fn advance(
conn: &Connection,
now: i64,
node_id: &str,
rrule: &str,
do_date: Option<i64>,
) -> Result<()> {
let anchor = do_date.unwrap_or(now);
match recurrence::next_occurrence(rrule, anchor, now)? {
Some(next) => {
conn.execute(
"UPDATE tasks SET do_date = ?1, state = 'outstanding' WHERE node_id = ?2",
(next, node_id),
)?;
}
None => {
conn.execute(
"UPDATE tasks SET state = 'done' WHERE node_id = ?1",
[node_id],
)?;
}
}
Ok(())
}
/// Skip the current occurrence of a recurring task: advance the do-date the same
/// way as completion but **without** logging a completion (tech-spec §4.4).
pub(super) fn skip(conn: &Connection, now: i64, node_id: &str) -> Result<Task> {
let task = require(conn, node_id)?;
let rrule = task
.recurrence
.as_deref()
.ok_or_else(|| Error::Integrity(format!("skip on non-recurring task {node_id}")))?;
advance(conn, now, node_id, rrule, task.do_date)?;
nodes::touch(conn, now, node_id)?;
require(conn, node_id)
}
/// The Tactical "what is next?" ranking for `owner` at `now` (tech-spec §7).
pub(super) fn next(
conn: &Connection,
owner: &str,
now: i64,
scope: Option<&str>,
limit: usize,
) -> Result<Vec<RankedTask>> {
let candidates = load_candidates(conn, owner)?;
Ok(ranking::rank(candidates, now, scope, limit))
}
/// Enumerate outstanding committed tasks for the Organizational view (the whole
/// set incl. backlog, tech-spec §6). Optional `scope` (project) and `attention`
/// filters; `include_blue` keeps on-deck items (default true for `list`).
pub(super) fn list(
conn: &Connection,
owner: &str,
scope: Option<&str>,
attention: Option<Attention>,
include_blue: bool,
) -> Result<Vec<Task>> {
let sql = "
SELECT t.node_id, t.attention, t.do_date, t.late_on, t.state, t.recurrence,
(SELECT dst_id FROM links
WHERE src_id = t.node_id AND type = 'in-project' AND tombstoned = 0
ORDER BY created_at, id LIMIT 1) AS project_id
FROM tasks t JOIN nodes n ON n.id = t.node_id
WHERE n.owner_id = ?1 AND n.tombstoned = 0 AND t.state = 'outstanding'
ORDER BY n.created_at, n.id";
let mut stmt = conn.prepare(sql)?;
let rows = stmt.query_map([owner], |row| {
let task = from_row(row)?;
let project: Option<String> = row.get("project_id")?;
Ok((task, project))
})?;
let mut out = Vec::new();
for row in rows {
let (task, project) = row?;
if let Some(s) = scope {
if project.as_deref() != Some(s) {
continue;
}
}
if let Some(a) = attention {
if task.attention != Some(a) {
continue;
}
}
if !include_blue && task.attention == Some(Attention::Blue) {
continue;
}
out.push(task);
}
Ok(out)
}
/// Working-set health counts (tech-spec §7) — surfaced honestly.
pub(super) fn health(conn: &Connection, owner: &str) -> Result<Health> {
let mut stmt = conn.prepare(
"SELECT t.attention FROM tasks t JOIN nodes n ON n.id = t.node_id
WHERE n.owner_id = ?1 AND n.tombstoned = 0 AND t.state = 'outstanding'",
)?;
let attentions: Vec<Option<String>> = stmt
.query_map([owner], |r| r.get(0))?
.collect::<rusqlite::Result<Vec<_>>>()?;
let mut orange_count = 0;
let mut active_count = 0;
let mut on_deck_count = 0;
for a in attentions.iter().flatten() {
match a.as_str() {
"orange" => {
orange_count += 1;
active_count += 1;
}
"red" | "white" => active_count += 1,
"blue" => on_deck_count += 1,
_ => {}
}
}
Ok(Health {
orange_count,
active_count,
on_deck_count,
conflict_count: 0,
sync_status: "local".to_string(),
})
}
/// Load every non-tombstoned committed task for `owner` as a ranking candidate,
/// joining in its project and canonical-context link targets.
fn load_candidates(conn: &Connection, owner: &str) -> Result<Vec<RankedTask>> {
let sql = "
SELECT n.id, n.title, n.created_at, n.tombstoned,
t.attention, t.do_date, t.late_on, t.state,
(SELECT dst_id FROM links
WHERE src_id = n.id AND type = 'in-project' AND tombstoned = 0
ORDER BY created_at, id LIMIT 1) AS project_id,
(SELECT dst_id FROM links
WHERE src_id = n.id AND type = 'canonical-context' AND tombstoned = 0
ORDER BY created_at, id LIMIT 1) AS ctx_id
FROM tasks t JOIN nodes n ON n.id = t.node_id
WHERE n.owner_id = ?1 AND n.tombstoned = 0";
let mut stmt = conn.prepare(sql)?;
let rows = stmt.query_map([owner], |row| {
let attention = match row.get::<_, Option<String>>("attention")? {
Some(s) => Some(
Attention::parse(&s)
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?,
),
None => None,
};
Ok(RankedTask {
node_id: row.get("id")?,
title: row.get("title")?,
attention,
do_date: row.get("do_date")?,
late_on: row.get("late_on")?,
state: TaskState::parse(&row.get::<_, String>("state")?)
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?,
tombstoned: row.get::<_, i64>("tombstoned")? != 0,
project_id: row.get("project_id")?,
canonical_context_id: row.get("ctx_id")?,
created_at: row.get("created_at")?,
})
})?;
Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}
/// Set a task's attention-state.
pub(super) fn set_attention(
conn: &Connection,
now: i64,
node_id: &str,
attention: Attention,
) -> Result<Task> {
let updated = conn.execute(
"UPDATE tasks SET attention = ?1 WHERE node_id = ?2",
(attention.as_str(), node_id),
)?;
if updated == 0 {
return Err(Error::NodeNotFound(node_id.to_string()));
}
nodes::touch(conn, now, node_id)?;
require(conn, node_id)
}