use anyhow::{Context, Result};
use chrono::{DateTime, Duration, Months, Utc};
use serde_json::{json, Value};
use crate::server::KhiveMcpServer;
use crate::tools::request::RequestParams;
use khive_runtime::{KhiveRuntime, Namespace};
use khive_storage::types::{SqlStatement, SqlValue};
const STALE_FIRING_TIMEOUT_MICROS: i64 = 5 * 60 * 1_000_000;
const DEFAULT_FIRE_GRACE_SECS: i64 = 300;
fn fire_grace_from_env() -> Duration {
let secs = std::env::var("KHIVE_FIRE_GRACE_SECS")
.ok()
.and_then(|s| s.parse::<i64>().ok())
.filter(|&s| s >= 0)
.unwrap_or(DEFAULT_FIRE_GRACE_SECS);
Duration::seconds(secs)
}
#[derive(Debug, Default)]
pub struct DrainSummary {
pub scanned: u64,
pub fired: u64,
pub advanced: u64,
pub failed: u64,
pub skipped_not_due: u64,
pub skipped_race: u64,
pub reclaimed: u64,
pub missed: Vec<uuid::Uuid>,
}
pub async fn run_pending_events(
db: Option<&str>,
namespace: &str,
verbose: bool,
) -> Result<DrainSummary> {
let ns = Namespace::parse(namespace)
.map_err(|e| anyhow::anyhow!("pending-events: invalid namespace {namespace:?}: {e}"))?;
let args = crate::args::Args {
db: db.map(str::to_string),
actor: None,
namespace: None,
no_embed: false,
pack: Vec::new(),
config: None,
daemon: false,
transport: None,
bind: None,
brain_profile: None,
resumed_generation: None,
};
let (server, schedule_rt) =
crate::serve::build_server_with_explicit_namespace(&args, ns, true, false)
.map_err(|e| anyhow::anyhow!("pending-events: build server: {e}"))?;
let rt = schedule_rt.ok_or_else(|| {
anyhow::anyhow!(
"pending-events: resolved pack set does not include \"schedule\"; nothing to drain"
)
})?;
run_pending_events_on(&rt, &server, verbose).await
}
pub async fn run_pending_events_on(
rt: &KhiveRuntime,
server: &KhiveMcpServer,
verbose: bool,
) -> Result<DrainSummary> {
let now = Utc::now();
let grace = fire_grace_from_env();
let mut summary = DrainSummary::default();
let stale_before = now
.timestamp_micros()
.saturating_sub(STALE_FIRING_TIMEOUT_MICROS);
summary.reclaimed = reclaim_stale_firing_events(rt, stale_before).await?;
if verbose && summary.reclaimed > 0 {
eprintln!(
"[pending-events] reclaimed {} stale \"firing\" row(s) back to \"pending\"",
summary.reclaimed
);
}
let namespaces = discover_pending_namespaces(rt, now).await?;
if verbose {
eprintln!(
"[pending-events] scan: now={}, namespaces_with_pending={}",
now.to_rfc3339(),
namespaces.len()
);
}
for ns_str in &namespaces {
if let Err(e) = Namespace::parse(ns_str) {
if verbose {
eprintln!("[pending-events] skip invalid namespace {ns_str:?}: {e}");
}
continue;
}
const PAGE_SIZE: u32 = 200;
let now_rfc = now.to_rfc3339();
let mut cursor: Option<(i64, String)> = None;
loop {
let (sql, params): (String, Vec<SqlValue>) = match &cursor {
None => (
"SELECT id, properties, created_at FROM notes \
WHERE namespace = ?1 AND kind = 'scheduled_event' \
AND deleted_at IS NULL \
AND json_extract(properties, '$.status') = 'pending' \
AND ( \
datetime(json_extract(properties, '$.trigger_at')) <= datetime(?2) \
OR datetime(json_extract(properties, '$.trigger_at')) IS NULL \
) \
ORDER BY created_at ASC, id ASC LIMIT ?3"
.to_string(),
vec![
SqlValue::Text(ns_str.clone()),
SqlValue::Text(now_rfc.clone()),
SqlValue::Integer(i64::from(PAGE_SIZE)),
],
),
Some((c_created_at, c_id)) => (
"SELECT id, properties, created_at FROM notes \
WHERE namespace = ?1 AND kind = 'scheduled_event' \
AND deleted_at IS NULL \
AND json_extract(properties, '$.status') = 'pending' \
AND ( \
datetime(json_extract(properties, '$.trigger_at')) <= datetime(?2) \
OR datetime(json_extract(properties, '$.trigger_at')) IS NULL \
) \
AND (created_at > ?3 OR (created_at = ?3 AND id > ?4)) \
ORDER BY created_at ASC, id ASC LIMIT ?5"
.to_string(),
vec![
SqlValue::Text(ns_str.clone()),
SqlValue::Text(now_rfc.clone()),
SqlValue::Integer(*c_created_at),
SqlValue::Text(c_id.clone()),
SqlValue::Integer(i64::from(PAGE_SIZE)),
],
),
};
let rows = {
let mut reader = rt
.sql()
.reader()
.await
.context("pending-events: open SQL reader for candidate page")?;
reader
.query_all(SqlStatement {
sql,
params,
label: Some("pending_events_candidate_page".into()),
})
.await
.with_context(|| {
format!("pending-events: candidate page query failed for ns={ns_str}")
})?
};
let page_len = rows.len();
if page_len == 0 {
break;
}
for row in &rows {
let id_str = match row.get("id") {
Some(SqlValue::Text(s)) => s.clone(),
other => {
if verbose {
eprintln!(
"[pending-events] skip row with unexpected id column {other:?}"
);
}
continue;
}
};
let row_created_at = match row.get("created_at") {
Some(SqlValue::Integer(v)) => *v,
other => {
if verbose {
eprintln!(
"[pending-events] skip row {id_str}: unexpected created_at \
column {other:?}"
);
}
continue;
}
};
cursor = Some((row_created_at, id_str.clone()));
let id = match uuid::Uuid::parse_str(&id_str) {
Ok(u) => u,
Err(e) => {
if verbose {
eprintln!("[pending-events] skip row: unparseable id {id_str:?}: {e}");
}
continue;
}
};
let mut properties: Option<Value> = match row.get("properties") {
Some(SqlValue::Text(s)) => match serde_json::from_str(s) {
Ok(v) => Some(v),
Err(e) => {
if verbose {
eprintln!(
"[pending-events] skip note {id}: unparseable properties: {e}"
);
}
continue;
}
},
Some(SqlValue::Null) | None => None,
other => {
if verbose {
eprintln!(
"[pending-events] skip note {id}: unexpected properties column \
{other:?}"
);
}
continue;
}
};
summary.scanned += 1;
let trigger_at_str = properties
.as_ref()
.and_then(|p| p.get("trigger_at"))
.and_then(Value::as_str)
.unwrap_or("");
let trigger_at = match trigger_at_str.parse::<DateTime<Utc>>() {
Ok(ts) => ts,
Err(_) => {
if verbose {
eprintln!(
"[pending-events] skip note {id}: unparseable trigger_at {trigger_at_str:?}"
);
}
summary.skipped_not_due += 1;
continue;
}
};
if trigger_at > now {
summary.skipped_not_due += 1;
continue;
}
let overdue = now.signed_duration_since(trigger_at);
let is_missed = overdue > grace;
let event_type = properties
.as_ref()
.and_then(|p| p.get("event_type"))
.and_then(Value::as_str)
.unwrap_or("remind");
let action_dsl: Option<String> = if event_type == "schedule" && !is_missed {
properties
.as_ref()
.and_then(|p| p.get("payload"))
.and_then(Value::as_str)
.map(str::to_string)
} else {
None
};
let repeat = properties
.as_ref()
.and_then(|p| p.get("repeat"))
.and_then(Value::as_str)
.map(str::to_string);
let claimed_firing_at = match claim_pending_event(rt, ns_str, id).await {
Ok(c) => c,
Err(e) => {
if verbose {
eprintln!("[pending-events] claim failed for note {id}: {e}");
}
summary.failed += 1;
continue;
}
};
let Some(claimed_firing_at) = claimed_firing_at else {
if verbose {
eprintln!(
"[pending-events] skip note {id}: no longer pending (concurrent \
cancel or claim)"
);
}
summary.skipped_race += 1;
continue;
};
if is_missed {
if verbose {
eprintln!(
"[pending-events] note {id} overdue by {}s (grace {}s): marking \
missed, not dispatching",
overdue.num_seconds(),
grace.num_seconds()
);
}
let mut props = properties.clone().unwrap_or_else(|| json!({}));
props["missed_at"] = json!(now.timestamp_micros());
match advance_repeat_past_missed(&repeat, trigger_at, now) {
Some(next_at) => {
props["trigger_at"] = json!(next_at.to_rfc3339());
props["status"] = json!("pending");
}
None => {
props["status"] = json!("missed");
}
}
let updated_at = Utc::now().timestamp_micros();
match finalize_fired_event(
rt,
ns_str,
id,
&props,
updated_at,
claimed_firing_at,
)
.await
{
Ok(true) => {
summary.missed.push(id);
}
Ok(false) => {
if verbose {
eprintln!(
"[pending-events] finalize no-op for {id}: row no longer in \
\"firing\" state"
);
}
summary.failed += 1;
}
Err(e) => {
if verbose {
eprintln!("[pending-events] finalize failed for {id}: {e}");
}
summary.failed += 1;
}
}
continue;
}
if let Some(dsl) = &action_dsl {
let dispatch_result = dispatch_action(dsl, ns_str, server, verbose).await;
if let Err(e) = dispatch_result {
if verbose {
eprintln!("[pending-events] dispatch failed for note {id}: {e}");
}
summary.failed += 1;
}
}
let fired_at_rfc = Utc::now().to_rfc3339();
let mut props = properties.clone().unwrap_or_else(|| json!({}));
let updated_at;
match next_trigger_at(&repeat, trigger_at) {
Some(next_at) => {
props["trigger_at"] = json!(next_at.to_rfc3339());
props["status"] = json!("pending");
props["fired_at"] = json!(fired_at_rfc);
properties = Some(props);
updated_at = Utc::now().timestamp_micros();
summary.advanced += 1;
}
None => {
props["status"] = json!("fired");
props["fired_at"] = json!(fired_at_rfc);
properties = Some(props);
updated_at = Utc::now().timestamp_micros();
summary.fired += 1;
}
}
let final_props = properties.clone().unwrap_or_else(|| json!({}));
match finalize_fired_event(
rt,
ns_str,
id,
&final_props,
updated_at,
claimed_firing_at,
)
.await
{
Ok(true) => {}
Ok(false) => {
if verbose {
eprintln!(
"[pending-events] finalize no-op for {id}: row no longer in \
\"firing\" state"
);
}
summary.failed += 1;
if summary.fired > 0 {
summary.fired -= 1;
}
if summary.advanced > 0 {
summary.advanced -= 1;
}
}
Err(e) => {
if verbose {
eprintln!("[pending-events] finalize failed for {id}: {e}");
}
summary.failed += 1;
if summary.fired > 0 {
summary.fired -= 1;
}
if summary.advanced > 0 {
summary.advanced -= 1;
}
}
}
}
if page_len < PAGE_SIZE as usize {
break;
}
}
}
Ok(summary)
}
async fn claim_pending_event(
rt: &KhiveRuntime,
namespace: &str,
id: uuid::Uuid,
) -> Result<Option<i64>> {
let updated_at = Utc::now().timestamp_micros();
let mut writer = rt
.sql()
.writer()
.await
.map_err(|e| anyhow::anyhow!("pending-events: open SQL writer: {e}"))?;
let rows = writer
.execute(SqlStatement {
sql: "UPDATE notes \
SET properties = json_set( \
json_set(COALESCE(properties, '{}'), '$.status', 'firing'), \
'$.firing_at', ?1 \
), \
updated_at = ?1 \
WHERE id = ?2 \
AND namespace = ?3 \
AND kind = 'scheduled_event' \
AND deleted_at IS NULL \
AND json_extract(properties, '$.status') = 'pending'"
.to_string(),
params: vec![
SqlValue::Integer(updated_at),
SqlValue::Text(id.to_string()),
SqlValue::Text(namespace.to_string()),
],
label: Some("pending_events_claim_firing".into()),
})
.await
.map_err(|e| anyhow::anyhow!("pending-events: claim conditional update: {e}"))?;
Ok((rows == 1).then_some(updated_at))
}
async fn reclaim_stale_firing_events(rt: &KhiveRuntime, stale_before_micros: i64) -> Result<u64> {
let mut writer = rt
.sql()
.writer()
.await
.map_err(|e| anyhow::anyhow!("pending-events: open SQL writer: {e}"))?;
let rows = writer
.execute(SqlStatement {
sql: "UPDATE notes \
SET properties = json_set(properties, '$.status', 'pending') \
WHERE kind = 'scheduled_event' \
AND deleted_at IS NULL \
AND json_extract(properties, '$.status') = 'firing' \
AND ( \
json_extract(properties, '$.firing_at') IS NULL \
OR CAST(json_extract(properties, '$.firing_at') AS INTEGER) < ?1 \
)"
.to_string(),
params: vec![SqlValue::Integer(stale_before_micros)],
label: Some("pending_events_reclaim_stale_firing".into()),
})
.await
.map_err(|e| anyhow::anyhow!("pending-events: reclaim stale firing rows: {e}"))?;
Ok(rows)
}
async fn finalize_fired_event(
rt: &KhiveRuntime,
namespace: &str,
id: uuid::Uuid,
properties: &Value,
updated_at: i64,
claimed_firing_at: i64,
) -> Result<bool> {
let mut properties = properties.clone();
if let Some(obj) = properties.as_object_mut() {
obj.remove("firing_at");
}
let props_json = serde_json::to_string(&properties)
.map_err(|e| anyhow::anyhow!("pending-events: serialize properties: {e}"))?;
let mut writer = rt
.sql()
.writer()
.await
.map_err(|e| anyhow::anyhow!("pending-events: open SQL writer: {e}"))?;
let rows = writer
.execute(SqlStatement {
sql: "UPDATE notes \
SET properties = ?1, updated_at = ?2 \
WHERE id = ?3 \
AND namespace = ?4 \
AND kind = 'scheduled_event' \
AND deleted_at IS NULL \
AND json_extract(properties, '$.status') = 'firing' \
AND CAST(json_extract(properties, '$.firing_at') AS INTEGER) = ?5"
.to_string(),
params: vec![
SqlValue::Text(props_json),
SqlValue::Integer(updated_at),
SqlValue::Text(id.to_string()),
SqlValue::Text(namespace.to_string()),
SqlValue::Integer(claimed_firing_at),
],
label: Some("pending_events_finalize_fired".into()),
})
.await
.map_err(|e| anyhow::anyhow!("pending-events: finalize conditional update: {e}"))?;
Ok(rows == 1)
}
fn next_trigger_at(repeat: &Option<String>, current: DateTime<Utc>) -> Option<DateTime<Utc>> {
match repeat.as_deref() {
Some("daily") => Some(current + Duration::days(1)),
Some("weekly") => Some(current + Duration::weeks(1)),
Some("monthly") => {
current.checked_add_months(Months::new(1))
}
Some(expr) if is_five_field_cron(expr) => {
tracing::warn!(
repeat = expr,
"pending-events: cron repeat expression cannot be advanced (not yet supported); \
event will be marked fired (one-shot)"
);
None
}
_ => None,
}
}
fn is_five_field_cron(expr: &str) -> bool {
expr.split_whitespace().count() == 5
}
fn advance_repeat_past_missed(
repeat: &Option<String>,
current: DateTime<Utc>,
now: DateTime<Utc>,
) -> Option<DateTime<Utc>> {
let mut current = current;
loop {
let next = next_trigger_at(repeat, current)?;
if next > now {
return Some(next);
}
current = next;
}
}
async fn dispatch_action(
action_dsl: &str,
namespace: &str,
server: &KhiveMcpServer,
verbose: bool,
) -> Result<()> {
let parsed = khive_request::parse_request(action_dsl).map_err(|e| {
anyhow::anyhow!("pending-events: action DSL parse error ({e}): {action_dsl:?}")
})?;
let mut ops_json: Vec<Value> = Vec::with_capacity(parsed.ops.len());
for op in &parsed.ops {
let mut args = serde_json::Map::new();
for (k, v) in &op.args {
let khive_request::ArgValue::Value(val) = v else {
return Err(anyhow::anyhow!(
"pending-events: non-literal scheduled action argument {k:?} is not \
replayable: {action_dsl:?}"
));
};
args.insert(k.clone(), val.clone());
}
args.insert(
"namespace".to_string(),
Value::String(namespace.to_string()),
);
ops_json.push(json!({ "tool": op.tool, "args": Value::Object(args) }));
}
let ops_str = serde_json::to_string(&ops_json)
.map_err(|e| anyhow::anyhow!("pending-events: serialize ops: {e}"))?;
if verbose {
eprintln!("[pending-events] dispatch ns={namespace}: {ops_str}");
}
let result = server
.dispatch_request_local(RequestParams {
ops: ops_str,
presentation: None,
presentation_per_op: None,
save_to: None,
format: None,
format_per_op: None,
})
.await
.map_err(|e| anyhow::anyhow!("pending-events: dispatch error: {e}"))?;
let parsed_result: Value = serde_json::from_str(&result).unwrap_or(Value::Null);
if let Some(results) = parsed_result.get("results").and_then(Value::as_array) {
let failures: Vec<_> = results
.iter()
.filter(|r| r.get("ok").and_then(Value::as_bool) == Some(false))
.collect();
if !failures.is_empty() {
let errs: Vec<String> = failures
.iter()
.filter_map(|r| r.get("error").and_then(Value::as_str).map(str::to_string))
.collect();
return Err(anyhow::anyhow!(
"pending-events: action produced {} failure(s): {}",
failures.len(),
errs.join("; ")
));
}
}
Ok(())
}
async fn discover_pending_namespaces(rt: &KhiveRuntime, now: DateTime<Utc>) -> Result<Vec<String>> {
use khive_storage::types::{SqlStatement, SqlValue};
let sql_access = rt.sql();
let mut reader = sql_access
.reader()
.await
.context("pending-events: open SQL reader")?;
let now_rfc = now.to_rfc3339();
let rows = reader
.query_all(SqlStatement {
sql: "SELECT DISTINCT namespace \
FROM notes \
WHERE kind = 'scheduled_event' \
AND deleted_at IS NULL \
AND json_extract(properties, '$.status') = 'pending' \
AND ( \
datetime(json_extract(properties, '$.trigger_at')) <= datetime(?1) \
OR datetime(json_extract(properties, '$.trigger_at')) IS NULL \
)"
.into(),
params: vec![SqlValue::Text(now_rfc)],
label: Some("pending_events_namespaces".into()),
})
.await
.context("pending-events: discover namespaces query")?;
let namespaces: Vec<String> = rows
.into_iter()
.filter_map(|row| {
row.get("namespace").and_then(|v| {
if let SqlValue::Text(s) = v {
Some(s.clone())
} else {
None
}
})
})
.collect();
Ok(namespaces)
}
pub fn print_summary(summary: &DrainSummary) {
let json = json!({
"scanned": summary.scanned,
"fired": summary.fired,
"advanced": summary.advanced,
"failed": summary.failed,
"skipped_not_due": summary.skipped_not_due,
"skipped_race": summary.skipped_race,
"reclaimed": summary.reclaimed,
"missed_count": summary.missed.len(),
"missed_ids": summary.missed.iter().map(uuid::Uuid::to_string).collect::<Vec<_>>(),
});
println!(
"{}",
serde_json::to_string_pretty(&json).expect("serialize")
);
}
const DEFAULT_TICK_INTERVAL_SECS: u64 = 60;
pub fn tick_interval_from_env() -> std::time::Duration {
let secs = std::env::var("KHIVE_SCHEDULE_TICK_SECS")
.ok()
.and_then(|s| s.parse::<u64>().ok())
.filter(|&s| s > 0)
.unwrap_or(DEFAULT_TICK_INTERVAL_SECS);
std::time::Duration::from_secs(secs)
}
pub async fn schedule_tick_loop(
rt: KhiveRuntime,
server: KhiveMcpServer,
interval: std::time::Duration,
) {
let mut ticker = tokio::time::interval_at(tokio::time::Instant::now() + interval, interval);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
ticker.tick().await;
match run_pending_events_on(&rt, &server, false).await {
Ok(summary) => {
if summary.fired > 0
|| summary.advanced > 0
|| summary.failed > 0
|| !summary.missed.is_empty()
{
tracing::info!(
scanned = summary.scanned,
fired = summary.fired,
advanced = summary.advanced,
missed = summary.missed.len(),
failed = summary.failed,
reclaimed = summary.reclaimed,
"schedule tick: drain pass complete"
);
}
}
Err(e) => {
tracing::warn!(error = %e, "schedule tick: drain pass failed");
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::FixedOffset;
use khive_runtime::RuntimeConfig;
use khive_storage::types::PageRequest;
use tempfile::NamedTempFile;
fn tmp_db() -> (NamedTempFile, String) {
let f = NamedTempFile::new().expect("tempfile");
let path = f.path().to_str().expect("utf8 path").to_string();
(f, path)
}
fn due_rfc3339() -> String {
(Utc::now() - Duration::seconds(5)).to_rfc3339()
}
fn now_rfc3339_for_ordering_check() -> String {
Utc::now().to_rfc3339()
}
async fn make_rt(db_path: &str) -> KhiveRuntime {
let cfg = RuntimeConfig {
db_path: Some(std::path::PathBuf::from(db_path)),
default_namespace: Namespace::parse("local").unwrap(),
embedding_model: None,
additional_embedding_models: vec![],
..Default::default()
};
KhiveRuntime::new(cfg).expect("runtime")
}
async fn drain_for_test(db_path: &str) -> Result<DrainSummary> {
let rt = make_rt(db_path).await;
let server = KhiveMcpServer::new(rt.clone()).map_err(|e| anyhow::anyhow!("{e}"))?;
run_pending_events_on(&rt, &server, false).await
}
async fn create_scheduled_event(
rt: &KhiveRuntime,
namespace: &str,
trigger_at: &str,
action_dsl: Option<&str>,
repeat: Option<&str>,
event_type: &str,
) -> uuid::Uuid {
let props = json!({
"trigger_at": trigger_at,
"repeat": repeat,
"status": "pending",
"event_type": event_type,
"payload": action_dsl,
"fired_at": null,
"cancelled_at": null,
});
let ns = Namespace::parse(namespace).expect("ns");
let token = rt.authorize(ns).expect("authorize");
let content = action_dsl.unwrap_or("test reminder");
let note = rt
.create_note(
&token,
"scheduled_event",
None,
content,
None,
Some(props),
vec![],
)
.await
.expect("create_note");
note.id
}
async fn get_note_props(rt: &KhiveRuntime, id: uuid::Uuid) -> Value {
let ns = Namespace::parse("local").unwrap();
let token = rt.authorize(ns).expect("authorize");
let store = rt.notes(&token).expect("notes");
let note = store
.get_note(id)
.await
.expect("get_note")
.expect("note exists");
note.properties.unwrap_or(json!({}))
}
#[tokio::test]
async fn due_event_is_fired() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let past = due_rfc3339();
let id =
create_scheduled_event(&rt, "local", &past, Some("stats()"), None, "schedule").await;
let summary = drain_for_test(&db_path).await.expect("drain");
assert!(summary.scanned >= 1, "must have scanned the due event");
assert!(
summary.fired >= 1 || summary.advanced >= 1,
"must fire or advance"
);
let props = get_note_props(&rt, id).await;
let status = props["status"].as_str().unwrap_or("");
assert!(
status == "fired" || status == "pending",
"status must be fired or pending (repeat), got {status:?}"
);
}
#[tokio::test]
async fn future_event_is_skipped() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let future = "2099-01-01T00:00:00Z";
let id =
create_scheduled_event(&rt, "local", future, Some("stats()"), None, "schedule").await;
let summary = drain_for_test(&db_path).await.expect("drain");
assert_eq!(summary.fired, 0, "future event must not be fired");
assert_eq!(summary.advanced, 0, "future event must not be advanced");
let props = get_note_props(&rt, id).await;
assert_eq!(
props["status"].as_str(),
Some("pending"),
"future event must remain pending"
);
}
#[tokio::test]
async fn due_event_with_positive_offset_trigger_at_fires() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let trigger_instant = Utc::now() - Duration::seconds(10);
let plus_four = FixedOffset::east_opt(4 * 3600).expect("valid offset");
let trigger_at = trigger_instant.with_timezone(&plus_four).to_rfc3339();
assert!(
trigger_at.as_str() > now_rfc3339_for_ordering_check().as_str(),
"test setup: {trigger_at:?} must sort AFTER a UTC now-string as raw text \
for this to exercise the lexicographic-ordering bug"
);
let id =
create_scheduled_event(&rt, "local", &trigger_at, Some("stats()"), None, "schedule")
.await;
let summary = drain_for_test(&db_path).await.expect("drain");
assert!(
summary.fired >= 1 || summary.advanced >= 1,
"a due event stored with a positive offset must still fire, got {summary:?}"
);
let props = get_note_props(&rt, id).await;
let status = props["status"].as_str().unwrap_or("");
assert!(
status == "fired" || status == "pending",
"status must be fired or pending (repeat), got {status:?}"
);
}
#[tokio::test]
async fn future_event_with_negative_offset_trigger_at_is_not_fired() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let trigger_instant = Utc::now() + Duration::hours(2);
let minus_eight = FixedOffset::west_opt(8 * 3600).expect("valid offset");
let trigger_at = trigger_instant.with_timezone(&minus_eight).to_rfc3339();
assert!(
trigger_at.as_str() < now_rfc3339_for_ordering_check().as_str(),
"test setup: {trigger_at:?} must sort BEFORE a UTC now-string as raw text \
for this to exercise the false-positive path"
);
let id =
create_scheduled_event(&rt, "local", &trigger_at, Some("stats()"), None, "schedule")
.await;
let summary = drain_for_test(&db_path).await.expect("drain");
assert_eq!(
summary.fired, 0,
"a chronologically future event must not be fired, got {summary:?}"
);
assert_eq!(
summary.advanced, 0,
"a chronologically future event must not be advanced, got {summary:?}"
);
let props = get_note_props(&rt, id).await;
assert_eq!(
props["status"].as_str(),
Some("pending"),
"future event must remain pending"
);
}
#[tokio::test]
async fn fired_event_is_idempotent() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let past = due_rfc3339();
let id =
create_scheduled_event(&rt, "local", &past, Some("stats()"), None, "schedule").await;
let s1 = drain_for_test(&db_path).await.expect("drain 1");
assert!(s1.scanned >= 1);
let s2 = drain_for_test(&db_path).await.expect("drain 2");
assert_eq!(s2.scanned, 0, "no pending events on second drain");
assert_eq!(s2.fired, 0, "no new fires on second drain");
let props = get_note_props(&rt, id).await;
let fired_at_1 = props["fired_at"].as_str().unwrap_or("").to_string();
assert!(
!fired_at_1.is_empty(),
"fired_at must be set after first drain"
);
let props2 = get_note_props(&rt, id).await;
assert_eq!(
props2["fired_at"].as_str().unwrap_or(""),
fired_at_1.as_str(),
"fired_at must not change on second drain"
);
}
#[tokio::test]
async fn daily_repeat_advances() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let past = due_rfc3339();
let id = create_scheduled_event(
&rt,
"local",
&past,
Some("stats()"),
Some("daily"),
"schedule",
)
.await;
let summary = drain_for_test(&db_path).await.expect("drain");
assert!(
summary.advanced >= 1,
"daily event must be advanced, not fired"
);
let props = get_note_props(&rt, id).await;
assert_eq!(
props["status"].as_str(),
Some("pending"),
"after advance, status must be pending"
);
let new_trigger = props["trigger_at"]
.as_str()
.expect("trigger_at must be set");
let new_ts: DateTime<Utc> = new_trigger.parse().expect("parseable ts");
let original: DateTime<Utc> = past.parse().unwrap();
assert_eq!(
new_ts,
original + Duration::days(1),
"daily advance must add 1 day"
);
}
#[tokio::test]
async fn namespace_isolation() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let ns_a = "ns-a";
let ns_b = "ns-b";
let past = due_rfc3339();
let id_a =
create_scheduled_event(&rt, ns_a, &past, Some("stats()"), None, "schedule").await;
let _id_b = create_scheduled_event(
&rt,
ns_b,
"2099-01-01T00:00:00Z",
Some("stats()"),
None,
"schedule",
)
.await;
let summary = drain_for_test(&db_path).await.expect("drain");
assert!(summary.scanned >= 1);
assert!(summary.fired >= 1 || summary.advanced >= 1);
let token_a = rt.authorize(Namespace::parse(ns_a).unwrap()).expect("auth");
let store_a = rt.notes(&token_a).expect("notes");
let note_a = store_a.get_note(id_a).await.expect("get").expect("exists");
let status_a = note_a
.properties
.as_ref()
.and_then(|p| p.get("status"))
.and_then(Value::as_str)
.unwrap_or("");
assert!(
status_a == "fired" || status_a == "pending",
"ns-a event must be fired or advanced, got {status_a:?}"
);
}
#[tokio::test]
async fn dispatch_failure_does_not_abort_drain() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let past = due_rfc3339();
let _id_bad = create_scheduled_event(
&rt,
"local",
&past,
Some("stats()"), None,
"schedule",
)
.await;
let id_bad2 = create_scheduled_event(
&rt,
"local",
&past,
Some("this_verb_does_not_exist(foo=\"bar\")"),
None,
"schedule",
)
.await;
let summary = drain_for_test(&db_path)
.await
.expect("drain must not abort");
assert!(summary.scanned >= 2, "both events must be scanned");
assert!(
summary.failed >= 1 || summary.fired >= 1,
"at least one event processed (failed or fired)"
);
let props_bad2 = get_note_props(&rt, id_bad2).await;
let _ = props_bad2["status"].as_str(); }
#[tokio::test]
#[serial_test::serial]
async fn replayable_action_dispatches_without_failure_at_trigger_time() {
struct RestoreTimeout(Option<String>);
impl Drop for RestoreTimeout {
fn drop(&mut self) {
match self.0.take() {
Some(v) => std::env::set_var("KHIVE_CHECKOUT_TIMEOUT_SECS", v),
None => std::env::remove_var("KHIVE_CHECKOUT_TIMEOUT_SECS"),
}
}
}
let _restore = RestoreTimeout(std::env::var("KHIVE_CHECKOUT_TIMEOUT_SECS").ok());
std::env::set_var("KHIVE_CHECKOUT_TIMEOUT_SECS", "120");
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let past = due_rfc3339();
let id = create_scheduled_event(
&rt,
"local",
&past,
Some("schedule.remind(content=\"ping\", at=\"2099-01-01T00:00:00Z\")"),
None,
"schedule",
)
.await;
let summary = drain_for_test(&db_path).await.expect("drain");
assert_eq!(
summary.failed, 0,
"a write-time-replayable action must dispatch cleanly at trigger time"
);
assert!(
summary.fired >= 1 || summary.advanced >= 1,
"the event must be processed"
);
let props = get_note_props(&rt, id).await;
assert_eq!(props["status"].as_str(), Some("fired"));
}
#[tokio::test]
async fn dispatch_action_rejects_non_literal_prev_reference() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let server = KhiveMcpServer::new(rt.clone()).expect("server");
let err = dispatch_action("stats() | get(id=$prev.id)", "local", &server, false)
.await
.unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("not replayable"),
"expected the specific non-literal-argument rejection message, got: {msg}"
);
}
#[tokio::test]
async fn dispatch_rejects_legacy_prev_reference_instead_of_dropping_it() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let past = due_rfc3339();
let _id = create_scheduled_event(
&rt,
"local",
&past,
Some("stats() | get(id=$prev.id)"),
None,
"schedule",
)
.await;
let summary = drain_for_test(&db_path)
.await
.expect("drain must not abort or panic on a legacy $prev row");
assert!(
summary.failed >= 1,
"a legacy $prev reference must surface as a dispatch failure, not a silent drop"
);
}
#[tokio::test]
async fn fire_claim_wins_race_against_concurrent_cancel() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let server = KhiveMcpServer::new(rt.clone()).expect("server");
let past = "2000-01-01T00:00:00Z";
let id =
create_scheduled_event(&rt, "local", past, Some("stats()"), None, "schedule").await;
let claimed_firing_at = claim_pending_event(&rt, "local", id)
.await
.expect("claim query")
.expect("claim must succeed on a fresh pending row");
let cancel_ops = serde_json::to_string(&serde_json::json!([
{ "tool": "schedule.cancel", "args": { "id": id.to_string() } }
]))
.expect("serialize cancel op");
let cancel_result = server
.dispatch_request_local(RequestParams {
ops: cancel_ops,
presentation: None,
presentation_per_op: None,
save_to: None,
format: None,
format_per_op: None,
})
.await
.expect("dispatch_request_local must not error at the RPC layer");
let cancel_json: Value = serde_json::from_str(&cancel_result).expect("valid JSON");
let op_result = &cancel_json["results"][0];
assert_eq!(
op_result["ok"], false,
"cancel of a claimed (firing) event must fail, not silently succeed: {cancel_json}"
);
let cancel_err = op_result["error"].as_str().unwrap_or("");
assert!(
cancel_err.contains("not pending"),
"cancel must report the event is no longer pending; got: {cancel_err}"
);
let finalized = finalize_fired_event(
&rt,
"local",
id,
&serde_json::json!({
"trigger_at": past,
"repeat": null,
"status": "fired",
"event_type": "schedule",
"payload": "stats()",
"fired_at": Utc::now().to_rfc3339(),
"cancelled_at": null,
}),
Utc::now().timestamp_micros(),
claimed_firing_at,
)
.await
.expect("finalize query");
assert!(
finalized,
"finalize must succeed on a row still in \"firing\""
);
let props = get_note_props(&rt, id).await;
assert_eq!(
props["status"].as_str().unwrap_or(""),
"fired",
"terminal state must be \"fired\"; cancel must not have won the race"
);
}
async fn force_set_properties(rt: &KhiveRuntime, id: uuid::Uuid, properties: &Value) {
let props_json = serde_json::to_string(properties).expect("serialize");
let mut writer = rt.sql().writer().await.expect("writer");
let rows = writer
.execute(SqlStatement {
sql: "UPDATE notes SET properties = ?1 WHERE id = ?2".to_string(),
params: vec![SqlValue::Text(props_json), SqlValue::Text(id.to_string())],
label: Some("test_force_set_properties".into()),
})
.await
.expect("force update");
assert_eq!(rows, 1, "test setup: row must exist");
}
#[tokio::test]
async fn stale_firing_row_is_reclaimed_and_fired() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let past = due_rfc3339();
let id =
create_scheduled_event(&rt, "local", &past, Some("stats()"), None, "schedule").await;
let stale_firing_at = Utc::now().timestamp_micros() - (STALE_FIRING_TIMEOUT_MICROS * 2);
force_set_properties(
&rt,
id,
&json!({
"trigger_at": past,
"repeat": null,
"status": "firing",
"event_type": "schedule",
"payload": "stats()",
"fired_at": null,
"cancelled_at": null,
"firing_at": stale_firing_at,
}),
)
.await;
let summary = drain_for_test(&db_path).await.expect("drain");
assert!(
summary.reclaimed >= 1,
"the stale firing row must be reclaimed, got summary={summary:?}"
);
assert!(
summary.fired >= 1 || summary.advanced >= 1,
"the reclaimed row must be fired (or advanced) in the same pass, \
got summary={summary:?}"
);
let props = get_note_props(&rt, id).await;
assert_eq!(
props["status"].as_str(),
Some("fired"),
"a reclaimed non-repeating event must end in \"fired\", got {props:?}"
);
}
#[tokio::test]
async fn fresh_firing_row_is_not_reclaimed() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let past = "2000-01-01T00:00:00Z";
let id =
create_scheduled_event(&rt, "local", past, Some("stats()"), None, "schedule").await;
let _claimed_firing_at = claim_pending_event(&rt, "local", id)
.await
.expect("claim query")
.expect("claim must succeed on a fresh pending row");
let summary = drain_for_test(&db_path).await.expect("drain");
assert_eq!(
summary.reclaimed, 0,
"a fresh firing row must not be reclaimed, got summary={summary:?}"
);
assert_eq!(
summary.fired, 0,
"a fresh firing row must not be fired by a drain pass that did not claim it"
);
let props = get_note_props(&rt, id).await;
assert_eq!(
props["status"].as_str(),
Some("firing"),
"a fresh firing row must remain firing (owned by the process that claimed it), \
got {props:?}"
);
}
#[tokio::test]
async fn stale_claimant_cannot_finalize_over_a_fresh_reclaim() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let past = "2000-01-01T00:00:00Z";
let id =
create_scheduled_event(&rt, "local", past, Some("stats()"), None, "schedule").await;
let a_claimed_firing_at = Utc::now().timestamp_micros() - (STALE_FIRING_TIMEOUT_MICROS * 2);
force_set_properties(
&rt,
id,
&json!({
"trigger_at": past,
"repeat": null,
"status": "firing",
"event_type": "schedule",
"payload": "stats()",
"fired_at": null,
"cancelled_at": null,
"firing_at": a_claimed_firing_at,
}),
)
.await;
let stale_before = Utc::now().timestamp_micros() - STALE_FIRING_TIMEOUT_MICROS;
let reclaimed = reclaim_stale_firing_events(&rt, stale_before)
.await
.expect("reclaim query");
assert_eq!(reclaimed, 1, "A's stale claim must be reclaimed");
let b_claimed_firing_at = claim_pending_event(&rt, "local", id)
.await
.expect("claim query")
.expect("B's claim must succeed on the reclaimed row");
assert_ne!(
a_claimed_firing_at, b_claimed_firing_at,
"B's claim token must differ from A's stale token"
);
let a_finalize_result = finalize_fired_event(
&rt,
"local",
id,
&json!({
"trigger_at": past,
"repeat": null,
"status": "fired",
"event_type": "schedule",
"payload": "stats()",
"fired_at": Utc::now().to_rfc3339(),
"cancelled_at": null,
}),
Utc::now().timestamp_micros(),
a_claimed_firing_at,
)
.await
.expect("finalize query must not error");
assert!(
!a_finalize_result,
"A's finalize with a stale claim token must be a no-op, not a successful write"
);
let props_after_a = get_note_props(&rt, id).await;
assert_eq!(
props_after_a["status"].as_str(),
Some("firing"),
"B's claim must survive A's stale finalize attempt untouched, got {props_after_a:?}"
);
assert_eq!(
props_after_a["firing_at"].as_i64(),
Some(b_claimed_firing_at),
"B's firing_at token must be unchanged by A's stale finalize attempt"
);
let b_finalize_result = finalize_fired_event(
&rt,
"local",
id,
&json!({
"trigger_at": past,
"repeat": null,
"status": "fired",
"event_type": "schedule",
"payload": "stats()",
"fired_at": Utc::now().to_rfc3339(),
"cancelled_at": null,
}),
Utc::now().timestamp_micros(),
b_claimed_firing_at,
)
.await
.expect("finalize query must not error");
assert!(
b_finalize_result,
"B's finalize with its own claim token must succeed"
);
let final_props = get_note_props(&rt, id).await;
assert_eq!(
final_props["status"].as_str(),
Some("fired"),
"terminal state must be \"fired\" via B's own claim, got {final_props:?}"
);
assert!(
final_props.get("firing_at").is_none() || final_props["firing_at"].is_null(),
"firing_at must be cleared on terminal finalize, got {final_props:?}"
);
}
#[tokio::test]
async fn cancel_on_stale_firing_row_still_fails_cleanly() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let server = KhiveMcpServer::new(rt.clone()).expect("server");
let past = "2000-01-01T00:00:00Z";
let id =
create_scheduled_event(&rt, "local", past, Some("stats()"), None, "schedule").await;
let stale_firing_at = Utc::now().timestamp_micros() - (STALE_FIRING_TIMEOUT_MICROS * 2);
force_set_properties(
&rt,
id,
&json!({
"trigger_at": past,
"repeat": null,
"status": "firing",
"event_type": "schedule",
"payload": "stats()",
"fired_at": null,
"cancelled_at": null,
"firing_at": stale_firing_at,
}),
)
.await;
let cancel_ops = serde_json::to_string(&serde_json::json!([
{ "tool": "schedule.cancel", "args": { "id": id.to_string() } }
]))
.expect("serialize cancel op");
let cancel_result = server
.dispatch_request_local(RequestParams {
ops: cancel_ops,
presentation: None,
presentation_per_op: None,
save_to: None,
format: None,
format_per_op: None,
})
.await
.expect("dispatch_request_local must not error at the RPC layer");
let cancel_json: Value = serde_json::from_str(&cancel_result).expect("valid JSON");
let op_result = &cancel_json["results"][0];
assert_eq!(
op_result["ok"], false,
"cancel of a stale-but-still-firing event must fail, not silently succeed \
(reclaim happens on drain, not cancel): {cancel_json}"
);
let cancel_err = op_result["error"].as_str().unwrap_or("");
assert!(
cancel_err.contains("not pending"),
"cancel must report the event is no longer pending; got: {cancel_err}"
);
let props = get_note_props(&rt, id).await;
assert_eq!(
props["status"].as_str().unwrap_or(""),
"firing",
"a failed cancel must not alter the row's status"
);
}
#[test]
fn next_trigger_at_daily() {
let base: DateTime<Utc> = "2026-06-01T09:00:00Z".parse().unwrap();
let next = next_trigger_at(&Some("daily".to_string()), base).unwrap();
assert_eq!(next, base + Duration::days(1));
}
#[test]
fn next_trigger_at_weekly() {
let base: DateTime<Utc> = "2026-06-01T09:00:00Z".parse().unwrap();
let next = next_trigger_at(&Some("weekly".to_string()), base).unwrap();
assert_eq!(next, base + Duration::weeks(1));
}
#[test]
fn next_trigger_at_monthly() {
let base: DateTime<Utc> = "2026-06-01T09:00:00Z".parse().unwrap();
let next = next_trigger_at(&Some("monthly".to_string()), base).unwrap();
let expected: DateTime<Utc> = "2026-07-01T09:00:00Z".parse().unwrap();
assert_eq!(next, expected);
}
#[test]
fn next_trigger_at_none_repeat_returns_none() {
let base: DateTime<Utc> = "2026-06-01T09:00:00Z".parse().unwrap();
assert!(next_trigger_at(&None, base).is_none());
}
#[test]
fn next_trigger_at_cron_returns_none() {
let base: DateTime<Utc> = "2026-06-01T09:00:00Z".parse().unwrap();
assert!(next_trigger_at(&Some("0 9 * * 1".to_string()), base).is_none());
}
#[test]
fn advance_repeat_past_missed_skips_all_accumulated_occurrences() {
let now: DateTime<Utc> = "2026-06-15T09:00:00Z".parse().unwrap();
let original: DateTime<Utc> = "2026-06-01T09:00:00Z".parse().unwrap();
let next = advance_repeat_past_missed(&Some("daily".to_string()), original, now).unwrap();
assert!(next > now, "advanced occurrence must be strictly future");
assert!(
next <= now + Duration::days(1),
"must land on the very next occurrence, not skip further than one interval past now"
);
assert_eq!(
next,
original + Duration::days(15),
"must be exactly the first daily occurrence after now (single advance, no burst)"
);
}
#[test]
fn advance_repeat_past_missed_no_repeat_returns_none() {
let now: DateTime<Utc> = "2026-06-15T09:00:00Z".parse().unwrap();
let original: DateTime<Utc> = "2026-06-01T09:00:00Z".parse().unwrap();
assert!(advance_repeat_past_missed(&None, original, now).is_none());
}
#[tokio::test]
async fn nine_overdue_events_beyond_grace_are_missed_with_zero_dispatch() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let past = "2000-01-01T00:00:00Z";
let marker = "nine-overdue-zero-dispatch-marker";
let action_dsl = format!("create(kind=\"observation\", content=\"{marker}\")");
let mut ids = Vec::new();
for _ in 0..9 {
let id = create_scheduled_event(
&rt,
"local",
past,
Some(action_dsl.as_str()),
None,
"schedule",
)
.await;
ids.push(id);
}
let summary = drain_for_test(&db_path).await.expect("drain");
assert_eq!(summary.scanned, 9, "all 9 overdue rows must be scanned");
assert_eq!(summary.fired, 0, "zero dispatches: nothing may be fired");
assert_eq!(
summary.advanced, 0,
"zero dispatches: nothing may be advanced"
);
assert_eq!(summary.failed, 0, "the missed path is not a failure");
assert_eq!(
summary.missed.len(),
9,
"all 9 overdue rows must be marked missed, got summary={summary:?}"
);
for id in &ids {
assert!(
summary.missed.contains(id),
"missed list must name every overdue id"
);
}
for id in ids {
let props = get_note_props(&rt, id).await;
assert_eq!(
props["status"].as_str(),
Some("missed"),
"note {id} must end in status=missed, got {props:?}"
);
assert!(
props["missed_at"].as_i64().is_some(),
"note {id} must have missed_at stamped, got {props:?}"
);
assert!(
props["fired_at"].is_null(),
"note {id} must never have fired_at set (never dispatched), got {props:?}"
);
}
let ns = Namespace::parse("local").unwrap();
let token = rt.authorize(ns).expect("authorize");
let store = rt.notes(&token).expect("notes");
let page = store
.query_notes(
"local",
Some("observation"),
PageRequest {
limit: 50,
offset: 0,
},
)
.await
.expect("query observation notes");
let marker_hits: Vec<_> = page.items.iter().filter(|n| n.content == marker).collect();
assert!(
marker_hits.is_empty(),
"the missed action must never dispatch: found {} marker note(s): {marker_hits:?}",
marker_hits.len()
);
}
#[tokio::test]
async fn overdue_within_grace_still_fires() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let trigger_at = (Utc::now() - Duration::seconds(60)).to_rfc3339();
let id =
create_scheduled_event(&rt, "local", &trigger_at, Some("stats()"), None, "schedule")
.await;
let summary = drain_for_test(&db_path).await.expect("drain");
assert!(
summary.missed.is_empty(),
"an event within grace must never be marked missed, got summary={summary:?}"
);
assert!(
summary.fired >= 1 || summary.advanced >= 1,
"an event within grace must be dispatched normally, got summary={summary:?}"
);
let props = get_note_props(&rt, id).await;
assert_eq!(
props["status"].as_str(),
Some("fired"),
"non-repeating in-grace event must end fired, got {props:?}"
);
assert!(
props["fired_at"].as_str().is_some(),
"in-grace event must have fired_at set, got {props:?}"
);
}
#[tokio::test]
async fn missed_repeat_is_rearmed_at_next_future_occurrence() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
let original_trigger: DateTime<Utc> = Utc::now() - Duration::days(10);
let id = create_scheduled_event(
&rt,
"local",
&original_trigger.to_rfc3339(),
Some("stats()"),
Some("daily"),
"schedule",
)
.await;
let summary = drain_for_test(&db_path).await.expect("drain");
assert_eq!(summary.fired, 0, "a missed repeat must not fire");
assert_eq!(
summary.advanced, 0,
"a missed repeat's re-arm is counted as missed, not advanced"
);
assert_eq!(
summary.missed.len(),
1,
"exactly one missed occurrence recorded"
);
assert!(summary.missed.contains(&id));
let props = get_note_props(&rt, id).await;
assert_eq!(
props["status"].as_str(),
Some("pending"),
"a missed repeat must be re-armed to pending, not left terminal, got {props:?}"
);
assert!(
props["missed_at"].as_i64().is_some(),
"missed_at must be stamped even though the row is re-armed, got {props:?}"
);
let new_trigger: DateTime<Utc> = props["trigger_at"]
.as_str()
.expect("trigger_at must be set")
.parse()
.expect("parseable trigger_at");
let now = Utc::now();
assert!(
new_trigger > now,
"re-armed trigger_at must be strictly in the future, got {new_trigger} (now={now})"
);
assert!(
new_trigger <= now + Duration::days(1),
"re-armed trigger_at must be the very next occurrence, not skip further \
(no catch-up burst), got {new_trigger} (now={now})"
);
}
#[tokio::test]
async fn backlog_larger_than_page_size_is_fully_drained_in_one_pass() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
const OVERDUE_ROW_COUNT: usize = 201; let past = "2000-01-01T00:00:00Z"; let mut ids = Vec::with_capacity(OVERDUE_ROW_COUNT);
for _ in 0..OVERDUE_ROW_COUNT {
let id =
create_scheduled_event(&rt, "local", past, Some("stats()"), None, "schedule").await;
ids.push(id);
}
let summary = drain_for_test(&db_path).await.expect("drain");
assert_eq!(
summary.scanned, OVERDUE_ROW_COUNT as u64,
"every overdue row across both pages must be scanned in one pass, got \
summary={summary:?}"
);
assert_eq!(
summary.missed.len(),
OVERDUE_ROW_COUNT,
"every overdue row across both pages must be marked missed in one pass \
(the page-boundary row must not be skipped), got summary={summary:?}"
);
for id in &ids {
assert!(
summary.missed.contains(id),
"missed list must name every row, including ones beyond the first page"
);
}
for id in ids {
let props = get_note_props(&rt, id).await;
assert_eq!(
props["status"].as_str(),
Some("missed"),
"note {id} must end in status=missed (not left pending past the page \
boundary), got {props:?}"
);
}
}
#[tokio::test]
async fn concurrent_drains_fire_each_row_exactly_once() {
let (_tmp, db_path) = tmp_db();
let rt = make_rt(&db_path).await;
const ROW_COUNT: usize = 20;
let past = due_rfc3339(); let mut ids = Vec::with_capacity(ROW_COUNT);
let mut markers = Vec::with_capacity(ROW_COUNT);
for i in 0..ROW_COUNT {
let marker = format!("concurrent-drain-marker-{i}");
let action_dsl = format!("create(kind=\"observation\", content=\"{marker}\")");
let id = create_scheduled_event(
&rt,
"local",
&past,
Some(action_dsl.as_str()),
None,
"schedule",
)
.await;
ids.push(id);
markers.push(marker);
}
let db_path_a = db_path.clone();
let db_path_b = db_path.clone();
let (summary_a, summary_b) = tokio::join!(
async move { drain_for_test(&db_path_a).await },
async move { drain_for_test(&db_path_b).await },
);
let summary_a = summary_a.expect("drain A");
let summary_b = summary_b.expect("drain B");
let total_dispatched =
summary_a.fired + summary_a.advanced + summary_b.fired + summary_b.advanced;
assert_eq!(
total_dispatched, ROW_COUNT as u64,
"every row must be dispatched exactly once across both concurrent drains, \
got a={summary_a:?} b={summary_b:?}"
);
assert_eq!(
summary_a.failed + summary_b.failed,
0,
"the CAS claim must make the losing drain skip cleanly (skipped_race), \
never fail: a={summary_a:?} b={summary_b:?}"
);
for id in &ids {
let props = get_note_props(&rt, *id).await;
assert_eq!(
props["status"].as_str(),
Some("fired"),
"note {id} must end fired exactly once, got {props:?}"
);
}
let ns = Namespace::parse("local").unwrap();
let token = rt.authorize(ns).expect("authorize");
let store = rt.notes(&token).expect("notes");
let page = store
.query_notes(
"local",
Some("observation"),
PageRequest {
limit: (ROW_COUNT as u32) + 10,
offset: 0,
},
)
.await
.expect("query observation notes");
for marker in &markers {
let hits: Vec<_> = page.items.iter().filter(|n| &n.content == marker).collect();
assert_eq!(
hits.len(),
1,
"marker {marker:?} must appear exactly once (double-dispatch check), \
found {}: {hits:?}",
hits.len()
);
}
}
struct SeatEnv {
original_cwd: std::path::PathBuf,
original_home: Option<std::ffi::OsString>,
_isolated_home: tempfile::TempDir,
}
impl SeatEnv {
fn enter(project_root: &std::path::Path) -> Self {
let original_cwd = std::env::current_dir().expect("read cwd");
let original_home = std::env::var_os("HOME");
let isolated_home = tempfile::tempdir().expect("isolated HOME tempdir");
std::env::set_current_dir(project_root).expect("chdir into seat project root");
std::env::set_var("HOME", isolated_home.path());
Self {
original_cwd,
original_home,
_isolated_home: isolated_home,
}
}
}
impl Drop for SeatEnv {
fn drop(&mut self) {
let _ = std::env::set_current_dir(&self.original_cwd);
match &self.original_home {
Some(h) => std::env::set_var("HOME", h),
None => std::env::remove_var("HOME"),
}
}
}
fn write_project_actor_config(project_root: &std::path::Path, actor_id: &str) {
std::fs::create_dir_all(project_root.join(".khive")).expect("mkdir .khive");
std::fs::write(
project_root.join(".khive/config.toml"),
format!("[actor]\nid = \"{actor_id}\"\n"),
)
.expect("write project actor config");
}
#[test]
#[serial_test::serial]
fn wrapper_seam_falls_through_to_project_actor_instead_of_clearing_it() {
std::env::remove_var("KHIVE_ACTOR");
std::env::remove_var("KHIVE_DB");
std::env::remove_var("KHIVE_PACKS");
std::env::remove_var("KHIVE_REQUIRE_ATTRIBUTED_ACTOR");
let seat_dir = tempfile::tempdir().expect("seat tempdir");
write_project_actor_config(seat_dir.path(), "lambda:pending-events-tenant");
let _seat_env = SeatEnv::enter(seat_dir.path());
let args = crate::args::Args {
db: Some(":memory:".to_string()),
actor: None,
namespace: None,
no_embed: false,
pack: Vec::new(),
config: None,
daemon: false,
transport: None,
bind: None,
brain_profile: None,
resumed_generation: None,
};
let ns = Namespace::parse("local").expect("local namespace");
let (_server, schedule_rt) =
crate::serve::build_server_with_explicit_namespace(&args, ns, true, false)
.expect("build_server_with_explicit_namespace must succeed");
let rt = schedule_rt.expect("\"schedule\" pack is in the default pack set");
assert_eq!(
rt.config().actor_id.as_deref(),
Some("lambda:pending-events-tenant"),
"a default namespace resolving to \"local\" must fall through to the \
project-configured [actor] id, not clear it as if it were an explicit \
--actor/--namespace override"
);
}
#[test]
#[serial_test::serial]
fn build_server_cli_seam_clears_actor_for_explicit_local_namespace() {
std::env::remove_var("KHIVE_ACTOR");
std::env::remove_var("KHIVE_DB");
std::env::remove_var("KHIVE_PACKS");
std::env::remove_var("KHIVE_REQUIRE_ATTRIBUTED_ACTOR");
let seat_dir = tempfile::tempdir().expect("seat tempdir");
write_project_actor_config(seat_dir.path(), "lambda:pending-events-tenant");
let _seat_env = SeatEnv::enter(seat_dir.path());
let args = crate::args::Args {
db: Some(":memory:".to_string()),
actor: None,
namespace: Some("local".to_string()),
no_embed: false,
pack: Vec::new(),
config: None,
daemon: false,
transport: None,
bind: None,
brain_profile: None,
resumed_generation: None,
};
let (_server, schedule_rt) =
crate::serve::build_server(&args).expect("build_server must succeed");
let rt = schedule_rt.expect("\"schedule\" pack is in the default pack set");
assert_eq!(
rt.config().actor_id,
None,
"build_server's genuine CLI-flag seam must still treat a present --namespace \
value as an explicit actor override and clear the actor for \"local\" — this \
is correct CLI behavior, unaffected by the wrapper-seam fix"
);
}
#[tokio::test]
#[serial_test::serial]
async fn wrapper_succeeds_under_strict_actor_mode_with_configured_project_actor() {
std::env::remove_var("KHIVE_ACTOR");
std::env::remove_var("KHIVE_DB");
std::env::remove_var("KHIVE_PACKS");
let prev_strict = std::env::var("KHIVE_REQUIRE_ATTRIBUTED_ACTOR").ok();
std::env::set_var("KHIVE_REQUIRE_ATTRIBUTED_ACTOR", "1");
let seat_dir = tempfile::tempdir().expect("seat tempdir");
write_project_actor_config(seat_dir.path(), "lambda:pending-events-tenant");
let _seat_env = SeatEnv::enter(seat_dir.path());
let result = run_pending_events(Some(":memory:"), "local", false).await;
match prev_strict {
Some(v) => std::env::set_var("KHIVE_REQUIRE_ATTRIBUTED_ACTOR", v),
None => std::env::remove_var("KHIVE_REQUIRE_ATTRIBUTED_ACTOR"),
}
result.expect(
"run_pending_events must succeed under strict actor mode when a project \
[actor] id is configured — the same config a live `kkernel mcp --daemon` \
boot in this project would resolve",
);
}
}