diff --git a/CHANGELOG.md b/CHANGELOG.md index b1e6d95a..82fc937d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ Cross-package release notes for relayburn. Package changelogs contain package-le ## [Unreleased] +- Read/report commands (`summary`, `hotspots`, `hotspots --findings`, and `sessions list`) now warn when the ledger has not received data within the configurable staleness threshold (24 hours by default); SDK and MCP consumers receive the same last-write timestamp and stale flag as data. - Pricing recognizes Claude 5 and GPT-5.6 models, prefers first-party tariffs over reseller duplicates, and applies long-context price tiers. - `burn hotspots --findings` surfaces unknown model pricing explicitly and ranks unpriced sessions by token volume instead of treating them as $0.00. diff --git a/README.md b/README.md index a8adab43..2d2e07cb 100644 --- a/README.md +++ b/README.md @@ -212,13 +212,14 @@ content/search data in `content.sqlite`. |---|---| | `~/.agentworkforce/burn/burn.sqlite` | Events, stamps, sessions, relationships, and archive metadata. | | `~/.agentworkforce/burn/content.sqlite` | Content blobs and the FTS5 search index. | -| `~/.agentworkforce/burn/config.json` | Content-storage and retention configuration. | +| `~/.agentworkforce/burn/config.json` | Content-storage, retention, and report-staleness configuration (`staleness.thresholdHours`; default `24`). | | `~/.agentworkforce/burn/pending-stamps/` | Temporary manifests used by launchers that do not expose a session ID before spawn. | | `RELAYBURN_HOME` | Override the whole Burn data directory. | | `RELAYBURN_SQLITE_PATH` | Override the events database path. | | `RELAYBURN_CONTENT_PATH` | Override the content database path. | -| `RELAYBURN_CONTENT_STORE=full|hash-only|off` | Control content sidecar storage. Default: `full`. | +| `RELAYBURN_CONTENT_STORE=full\|hash-only\|off` | Control content sidecar storage. Default: `full`. | | `RELAYBURN_CONTENT_TTL_DAYS=` | Sidecar retention. Default: `90`. | +| `RELAYBURN_STALE_AFTER_HOURS=` | Age after which reads warn that the ledger is stale. Default: `24`; set `-1` to disable. | Reports read local data from the ledger and derived sidecars. diff --git a/crates/relayburn-cli/src/commands/freshness.rs b/crates/relayburn-cli/src/commands/freshness.rs new file mode 100644 index 00000000..a2cc636f --- /dev/null +++ b/crates/relayburn-cli/src/commands/freshness.rs @@ -0,0 +1,18 @@ +use relayburn_sdk::LedgerFreshness; + +use crate::cli::GlobalArgs; + +/// Present SDK freshness data on stderr without coupling the SDK to a UI. +pub(crate) fn warn_if_stale(freshness: &LedgerFreshness, globals: &GlobalArgs) { + if !freshness.stale { + return; + } + let last = relayburn_cli::util::time::format_optional_epoch_ms(freshness.last_write_at_ms); + let threshold_hours = freshness.stale_after_ms.unwrap_or_default() as f64 / 3_600_000.0; + crate::render::ux::print_warning( + &format!( + "ledger data may be stale (last write: {last}; threshold: {threshold_hours:.1}h). If expected activity is missing, run `burn ingest` before relying on this report." + ), + globals, + ); +} diff --git a/crates/relayburn-cli/src/commands/hotspots/mod.rs b/crates/relayburn-cli/src/commands/hotspots/mod.rs index 95014fe7..bed21724 100644 --- a/crates/relayburn-cli/src/commands/hotspots/mod.rs +++ b/crates/relayburn-cli/src/commands/hotspots/mod.rs @@ -265,6 +265,7 @@ fn run_inner(globals: &GlobalArgs, args: HotspotsArgs) -> anyhow::Result { let raw_opts = progress.ingest_options(ledger_home.clone()); ingest_all(handle.raw_mut(), &raw_opts)?; } + let freshness = handle.ledger_freshness()?; drop(handle); let session_filter = match args.session.as_deref() { @@ -284,6 +285,7 @@ fn run_inner(globals: &GlobalArgs, args: HotspotsArgs) -> anyhow::Result { ledger_home, })?; progress.finish_and_clear(); + crate::commands::freshness::warn_if_stale(&freshness, globals); if globals.json { emit_json(&result)?; diff --git a/crates/relayburn-cli/src/commands/mcp_server.rs b/crates/relayburn-cli/src/commands/mcp_server.rs index 6332a8fd..a1dc57cc 100644 --- a/crates/relayburn-cli/src/commands/mcp_server.rs +++ b/crates/relayburn-cli/src/commands/mcp_server.rs @@ -268,8 +268,10 @@ impl Server { "Cheap polling primitive over the burn ledger. Returns \ `{count}:{maxMtimeUnix}:{totalBytes}` — three integers \ joined by colons. Clients keep the last-seen value and \ - skip re-querying when it's unchanged. Optionally scoped \ - to a session id or a project path. Read-only.", + skip re-querying when it's unchanged. The response also \ + includes ledgerFreshness; check ledgerFreshness.stale \ + before relying on ledger reads. Optionally scoped to a \ + session id or a project path. Read-only.", "inputSchema": { "type": "object", "properties": { @@ -352,23 +354,25 @@ impl Server { let handle_guard = self.handle.lock().await; let result = handle_guard.fingerprint(scope); + let freshness = handle_guard.ledger_freshness(); drop(handle_guard); let fp = match result { Ok(fp) => fp, Err(err) => { - write_success( - id, - json!({ - "content": [{ "type": "text", "text": err.to_string() }], - "isError": true, - }), - ); + write_tool_error(id, err.to_string()); + return; + } + }; + let freshness = match freshness { + Ok(value) => value, + Err(err) => { + write_tool_error(id, err.to_string()); return; } }; - let payload = json!({ "fingerprint": fp.as_str() }); + let payload = json!({ "fingerprint": fp.as_str(), "ledgerFreshness": freshness }); let text = serde_json::to_string(&payload).unwrap_or_else(|_| "{}".to_string()); write_success( id, @@ -404,22 +408,16 @@ impl Server { }; let handle_guard = self.handle.lock().await; let result = handle_guard.session_cost(opts); + let freshness = handle_guard.ledger_freshness(); drop(handle_guard); let mut payload: SessionCostResult = match result { Ok(r) => r, Err(err) => { - let msg = err.to_string(); // Per MCP convention: tool errors are non-throwing // results with `isError: true`. Reserve JSON-RPC errors // for protocol problems (parse / method-not-found). - write_success( - id, - json!({ - "content": [{ "type": "text", "text": msg }], - "isError": true, - }), - ); + write_tool_error(id, err.to_string()); return; } }; @@ -434,7 +432,18 @@ impl Server { Some("no session id provided and server was not registered with one".to_string()); } - let value = serde_json::to_value(&payload).unwrap_or(Value::Null); + let mut value = serde_json::to_value(&payload).unwrap_or(Value::Null); + match freshness { + Ok(freshness) => { + if let Some(object) = value.as_object_mut() { + object.insert("ledgerFreshness".to_string(), json!(freshness)); + } + } + Err(err) => { + write_tool_error(id, err.to_string()); + return; + } + } let text = serde_json::to_string(&value).unwrap_or_else(|_| "{}".to_string()); write_success( id, @@ -459,6 +468,16 @@ fn write_success(id: &Value, result: Value) { write_response(&serde_json::to_value(&env).unwrap_or(Value::Null)); } +fn write_tool_error(id: &Value, message: impl Into) { + write_success( + id, + json!({ + "content": [{ "type": "text", "text": message.into() }], + "isError": true, + }), + ); +} + fn error_envelope(id: &Value, code: i32, message: &str, data: Option) -> Value { let env = JsonRpcError { jsonrpc: "2.0", diff --git a/crates/relayburn-cli/src/commands/mod.rs b/crates/relayburn-cli/src/commands/mod.rs index 43b59b4f..47e39c11 100644 --- a/crates/relayburn-cli/src/commands/mod.rs +++ b/crates/relayburn-cli/src/commands/mod.rs @@ -20,6 +20,7 @@ pub mod compare; pub mod flow; +mod freshness; pub mod hotspots; pub mod ingest; pub mod mcp_server; diff --git a/crates/relayburn-cli/src/commands/sessions.rs b/crates/relayburn-cli/src/commands/sessions.rs index f5520fad..88c6c78b 100644 --- a/crates/relayburn-cli/src/commands/sessions.rs +++ b/crates/relayburn-cli/src/commands/sessions.rs @@ -73,7 +73,9 @@ fn run_list_inner(globals: &GlobalArgs, args: SessionsListArgs) -> anyhow::Resul let result = handle.sessions_list(sdk_opts).inspect_err(|_| { progress.finish_and_clear(); })?; + let freshness = handle.ledger_freshness()?; progress.finish_and_clear(); + crate::commands::freshness::warn_if_stale(&freshness, globals); if globals.json { emit_json( diff --git a/crates/relayburn-cli/src/commands/state.rs b/crates/relayburn-cli/src/commands/state.rs index 9c4974a3..b6faf079 100644 --- a/crates/relayburn-cli/src/commands/state.rs +++ b/crates/relayburn-cli/src/commands/state.rs @@ -136,6 +136,9 @@ fn format_status(s: &StateStatus) -> String { " last rebuild: {}\n", s.archive.last_rebuild_at.as_deref().unwrap_or("never") )); + let last_write = + relayburn_cli::util::time::format_optional_epoch_ms(s.archive.last_write_at_ms); + out.push_str(&format!(" last write: {last_write}\n")); out.push_str("config:\n"); out.push_str(&format!(" store: {}\n", s.config.store)); let retention = if s.config.retention_forever { diff --git a/crates/relayburn-cli/src/commands/summary/mod.rs b/crates/relayburn-cli/src/commands/summary/mod.rs index 28c5babf..2b2e5811 100644 --- a/crates/relayburn-cli/src/commands/summary/mod.rs +++ b/crates/relayburn-cli/src/commands/summary/mod.rs @@ -326,6 +326,7 @@ fn run_inner(globals: &GlobalArgs, args: SummaryArgs) -> anyhow::Result { include_quality: args.quality, ledger_home: None, }; + let freshness = handle.ledger_freshness()?; // `--bucket` switches to a per-bucket time-series of the grouped summary. // Parsing/validation already happened above, before the ledger was opened. @@ -337,6 +338,7 @@ fn run_inner(globals: &GlobalArgs, args: SummaryArgs) -> anyhow::Result { progress.finish_and_clear(); })?; progress.finish_and_clear(); + crate::commands::freshness::warn_if_stale(&freshness, globals); return emit_summary_timeseries(globals, &series, &ingest_report); } @@ -345,6 +347,7 @@ fn run_inner(globals: &GlobalArgs, args: SummaryArgs) -> anyhow::Result { progress.finish_and_clear(); })?; progress.finish_and_clear(); + crate::commands::freshness::warn_if_stale(&freshness, globals); match report { SummaryReport::Grouped(report) => { diff --git a/crates/relayburn-cli/src/util/time.rs b/crates/relayburn-cli/src/util/time.rs index c5b10d50..b7e8c3c0 100644 --- a/crates/relayburn-cli/src/util/time.rs +++ b/crates/relayburn-cli/src/util/time.rs @@ -38,6 +38,16 @@ pub fn iso_from_system_time(t: std::time::SystemTime) -> String { iso_from_ms(total_ms) } +/// Format an optional Unix millisecond timestamp for human status/warning +/// surfaces. Missing ledger history is rendered consistently as `never`. +pub fn format_optional_epoch_ms(value: Option) -> String { + value + .map(|ms| { + iso_from_system_time(std::time::UNIX_EPOCH + std::time::Duration::from_millis(ms)) + }) + .unwrap_or_else(|| "never".to_string()) +} + fn iso_from_ms(total_ms: i64) -> String { let total_secs = total_ms.div_euclid(1000); let ms = total_ms.rem_euclid(1000) as u32; @@ -102,4 +112,13 @@ mod tests { let t = UNIX_EPOCH + Duration::from_millis(0); assert_eq!(iso_from_system_time(t), "1970-01-01T00:00:00.000Z"); } + + #[test] + fn optional_epoch_ms_formats_value_and_missing() { + assert_eq!(format_optional_epoch_ms(None), "never"); + assert_eq!( + format_optional_epoch_ms(Some(1_234)), + "1970-01-01T00:00:01.234Z" + ); + } } diff --git a/crates/relayburn-cli/tests/golden.rs b/crates/relayburn-cli/tests/golden.rs index c2129ab7..342db2ce 100644 --- a/crates/relayburn-cli/tests/golden.rs +++ b/crates/relayburn-cli/tests/golden.rs @@ -274,6 +274,33 @@ fn normalize(text: &str, ledger_home: &Path, project_dir: &Path) -> String { out = squash_numeric_field(&out, "ledgerMtimeMsCurrent", "${MTIME}"); out = squash_numeric_field(&out, "lastBuiltAt", "${TS}"); out = squash_numeric_field(&out, "lastRebuildAt", "${TS}"); + out = squash_numeric_field(&out, "lastWriteAtMs", "${TS}"); + out = squash_line_value(&out, " last write:", "${TS}"); + out +} + +/// Replace the value after a human-output label while preserving the label's +/// padding. This keeps wall-clock fields deterministic in golden snapshots. +/// The `never` sentinel (a ledger with no write clock) is not a wall-clock +/// value and passes through unchanged, so goldens can distinguish it from a +/// real timestamp. +fn squash_line_value(text: &str, label: &str, placeholder: &str) -> String { + let mut out = String::with_capacity(text.len()); + for segment in text.split_inclusive('\n') { + let (line, newline) = segment + .strip_suffix('\n') + .map_or((segment, ""), |line| (line, "\n")); + match line.strip_prefix(label) { + Some(rest) if rest.trim() != "never" => { + let padding_len = rest.len() - rest.trim_start().len(); + out.push_str(label); + out.push_str(&rest[..padding_len]); + out.push_str(placeholder); + } + _ => out.push_str(line), + } + out.push_str(newline); + } out } @@ -380,7 +407,7 @@ fn tempdir_under(parent: &Path) -> PathBuf { #[cfg(test)] mod tests { - use super::squash_numeric_field; + use super::{squash_line_value, squash_numeric_field}; #[test] fn squash_numeric_field_matches_space_and_tab() { @@ -417,4 +444,19 @@ mod tests { let out = squash_numeric_field(input, "lastBuiltAt", "${TS}"); assert_eq!(out, input); } + + #[test] + fn squash_line_value_preserves_padding_and_trailing_newline() { + let input = "archive state:\n last write: 2026-08-03T04:00:00Z\nconfig:\n"; + assert_eq!( + squash_line_value(input, " last write:", "${TS}"), + "archive state:\n last write: ${TS}\nconfig:\n" + ); + } + + #[test] + fn squash_line_value_preserves_never_sentinel() { + let input = "archive state:\n last write: never\nconfig:\n"; + assert_eq!(squash_line_value(input, " last write:", "${TS}"), input); + } } diff --git a/crates/relayburn-cli/tests/smoke.rs b/crates/relayburn-cli/tests/smoke.rs index 46178f5e..8e768fe8 100644 --- a/crates/relayburn-cli/tests/smoke.rs +++ b/crates/relayburn-cli/tests/smoke.rs @@ -53,6 +53,85 @@ fn burn() -> Command { Command::cargo_bin("burn").expect("`burn` binary must build for the smoke test") } +fn burn_without_stale_threshold_env() -> Command { + let mut command = burn(); + command.env_remove("RELAYBURN_STALE_AFTER_HOURS"); + command +} + +fn seed_one_turn(home: &std::path::Path) { + let mut handle = relayburn_sdk::Ledger::open(relayburn_sdk::LedgerOpenOptions::with_home(home)) + .expect("open test ledger"); + let turn: relayburn_sdk::TurnRecord = serde_json::from_value(serde_json::json!({ + "v": 1, + "source": "codex", + "sessionId": "stale-session", + "messageId": "stale-message", + "turnIndex": 0, + "ts": "2026-01-01T00:00:00.000Z", + "model": "gpt-5.2-codex", + "usage": {"input": 1, "output": 1, "reasoning": 0, "cacheRead": 0, "cacheCreate5m": 0, "cacheCreate1h": 0}, + "toolCalls": [] + })) + .expect("deserialize test turn"); + handle.raw_mut().append_turns(&[turn]).expect("seed turn"); +} + +#[test] +fn stale_warning_is_uniform_across_requested_read_surface() { + let home = tempfile::TempDir::new().expect("tmp RELAYBURN_HOME"); + seed_one_turn(home.path()); + std::fs::write( + home.path().join("config.json"), + r#"{"staleness":{"thresholdHours":0}}"#, + ) + .unwrap(); + std::thread::sleep(std::time::Duration::from_millis(2)); + + for args in [ + vec!["summary"], + vec!["hotspots"], + vec!["hotspots", "--findings"], + vec!["sessions", "list", "--since", "12m"], + ] { + burn_without_stale_threshold_env() + .args(["--ledger-path", home.path().to_str().expect("utf-8 path")]) + .args(&args) + .assert() + .success() + .stderr(predicate::str::contains("ledger data may be stale")); + } +} + +#[test] +fn fresh_ledger_does_not_warn() { + let home = tempfile::TempDir::new().expect("tmp RELAYBURN_HOME"); + seed_one_turn(home.path()); + burn_without_stale_threshold_env() + .args([ + "--ledger-path", + home.path().to_str().expect("utf-8 path"), + "summary", + ]) + .assert() + .success() + .stderr(predicate::str::contains("ledger data may be stale").not()); +} + +#[test] +fn never_written_ledger_warns() { + let home = tempfile::TempDir::new().expect("tmp RELAYBURN_HOME"); + burn_without_stale_threshold_env() + .args([ + "--ledger-path", + home.path().to_str().expect("utf-8 path"), + "summary", + ]) + .assert() + .success() + .stderr(predicate::str::contains("ledger data may be stale")); +} + #[test] fn top_level_help_lists_every_subcommand() { let output = burn().arg("--help").assert().success().get_output().clone(); diff --git a/crates/relayburn-sdk-node/src/lib.rs b/crates/relayburn-sdk-node/src/lib.rs index bbfe033f..c72fd6c1 100644 --- a/crates/relayburn-sdk-node/src/lib.rs +++ b/crates/relayburn-sdk-node/src/lib.rs @@ -34,8 +34,9 @@ //! rather than dragging `chrono::DateTime` or `Date` through the FFI. //! Matches the public Node facade types. //! - **`async fn` SDK verbs → `Promise` on the JS side.** napi-rs's -//! `tokio_rt` feature drives this; we mark `ingest` `async fn` and the -//! sync verbs (`summary`, `sessionCost`, …) as plain `fn` returning +//! `tokio_rt` feature drives this; blocking operations such as `ingest` and +//! `ledgerFreshness` run through Tokio's blocking pool. Lightweight sync +//! verbs (`summary`, `sessionCost`, …) remain plain `fn` returning //! `Result`. //! - **Errors → typed `BurnError` JS class (sync verbs only).** Domain //! failures from the SDK (`anyhow::Error`) and argument-shape errors @@ -48,7 +49,7 @@ //! [`BurnErrorCode`] enum is exported as a `string_enum` so TS code //! can reference the codes by name without stringly-typed literals. //! -//! **Async exception — [`ingest`].** napi-rs 2.x's `async fn` lowering +//! **Async exception — [`ingest`] and [`ledger_freshness`].** napi-rs 2.x's `async fn` lowering //! in `napi-derive` runs through `napi::bindgen_prelude::execute_tokio_future` //! ([`napi-derive-backend`]'s `codegen/fn.rs`), which is hard-typed to //! `Result>` — and `Status` is a *closed* enum @@ -62,12 +63,12 @@ //! `crates/relayburn-sdk-node/src/lib.rs` git history for the //! evaluation. We deliberately don't pay that complexity in v1. //! -//! **Concrete contract for [`ingest`]:** the returned `Promise` -//! rejects with a JS `Error` whose `.code === 'GenericFailure'` and -//! whose `.message` is the rendered `anyhow::Error` chain from the -//! SDK. JS callers branching on `e.code` should match `'GenericFailure'` -//! for ingest failures (or, more robustly, gate on `e.message` -//! substrings if discrimination is required). A future PR can tighten +//! **Concrete contract for async bindings:** the returned promise rejects +//! with a JS `Error` whose `.code === 'GenericFailure'` and whose `.message` +//! is the rendered `anyhow::Error` chain from the SDK. JS callers branching +//! on `e.code` should match `'GenericFailure'` for these failures (or, more +//! robustly, gate on `e.message` substrings if discrimination is required). +//! A future PR can tighten //! this — likely by upgrading to napi-rs 3.x once the `string_enum` //! and `BigInt` ergonomics there are validated against the rest of //! the binding — at which point `e.code` becomes one of the @@ -90,7 +91,9 @@ use std::path::PathBuf; use std::ptr; use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use napi::bindgen_prelude::{BigInt, Error as NapiError, Result as NapiResult, ToNapiValue}; +use napi::bindgen_prelude::{ + BigInt, Either, Error as NapiError, Null, Result as NapiResult, ToNapiValue, +}; use napi::sys; use napi_derive::napi; use serde_json::Value as JsonValue; @@ -374,6 +377,41 @@ fn open_options(home: Option, content_home: Option) -> sdk::Ledg } } +#[napi(object)] +pub struct LedgerFreshnessOptions { + pub ledger_home: Option, +} + +#[napi(object)] +pub struct LedgerFreshness { + pub last_write_at_ms: Option, + pub stale_after_ms: Either, + pub stale: bool, +} + +/// Return the shared SDK staleness flag without printing. MCP and other Node +/// presenters can attach this data to their own response envelopes. +#[napi] +pub async fn ledger_freshness( + opts: Option, +) -> Result { + let home = opts.and_then(|o| o.ledger_home); + let freshness = tokio::task::spawn_blocking(move || { + let handle = sdk::Ledger::open(open_options(home, None))?; + handle.ledger_freshness() + }) + .await + .map_err(|e| NapiError::from_reason(format!("ledger freshness task panicked: {e}")))? + .map_err(|e| NapiError::from_reason(format!("{e:#}")))?; + Ok(LedgerFreshness { + last_write_at_ms: freshness.last_write_at_ms.map(|v| v as f64), + stale_after_ms: freshness + .stale_after_ms + .map_or(Either::B(Null), |v| Either::A(v as f64)), + stale: freshness.stale, + }) +} + // --------------------------------------------------------------------------- // writePendingStamp // --------------------------------------------------------------------------- diff --git a/crates/relayburn-sdk/src/ledger.rs b/crates/relayburn-sdk/src/ledger.rs index 702fc12c..5defb7d0 100644 --- a/crates/relayburn-sdk/src/ledger.rs +++ b/crates/relayburn-sdk/src/ledger.rs @@ -31,9 +31,12 @@ use std::path::PathBuf; use rusqlite::params; +#[cfg(test)] +pub(crate) use crate::ledger::config::CONFIG_ENV_LOCK; pub use crate::ledger::config::{ config_path, config_path_at_home, load_config, load_config_at, load_config_with_home, - BurnConfig, ContentConfig, Retention, DEFAULT_RETENTION_DAYS, + load_staleness_config, load_staleness_config_at, load_staleness_config_with_home, BurnConfig, + ContentConfig, Retention, StalenessConfig, DEFAULT_RETENTION_DAYS, DEFAULT_STALE_AFTER_HOURS, }; pub use crate::ledger::content::{PruneStats, SearchHit, SearchOptions}; pub use crate::ledger::error::{LedgerError, Result}; @@ -81,6 +84,19 @@ impl Ledger { &self.conns.content_path } + /// Wall-clock time of the most recent event/derived-row mutation in + /// `burn.sqlite`, in Unix milliseconds. Content-sidecar-only writes are + /// excluded because report freshness tracks ingested activity rather than + /// blob persistence. `None` means the ledger has never received activity. + pub fn last_write_at_ms(&self) -> Result> { + let value: Option = self.conns.burn.query_row( + "SELECT last_write_at_ms FROM archive_state WHERE id = 1", + [], + |row| row.get(0), + )?; + Ok(value.and_then(|v| u64::try_from(v).ok())) + } + // --- append paths ------------------------------------------------- pub fn append_turns(&mut self, turns: &[crate::reader::TurnRecord]) -> Result { @@ -350,8 +366,11 @@ impl Ledger { // --- state rebuild ----------------------------------------------- /// Drop the derivable tables in `burn.sqlite` and the entire - /// `content.sqlite`, then re-create them empty. Stamps, archive - /// state, and ingest cursors are preserved. + /// `content.sqlite`, then re-create them empty. Stamps and ingest cursors + /// are preserved; archive timestamps remain except for + /// `last_write_at_ms`, which is cleared until re-ingest writes derived + /// rows again. The source fingerprint is also cleared so re-ingest cannot + /// incorrectly short-circuit. /// /// Returns the path to the (now-empty) content DB so the caller can /// move on to re-ingest from upstream files. Re-ingest is the @@ -395,7 +414,8 @@ impl Ledger { // this only forces the next ingest to re-examine source state rather // than trusting a fingerprint recorded before the drop. self.conns.burn.execute( - "UPDATE archive_state SET last_rebuild_at = ?, source_fingerprint = '' WHERE id = 1", + "UPDATE archive_state SET last_rebuild_at = ?, source_fingerprint = '', \ + last_write_at_ms = NULL WHERE id = 1", params![now], )?; @@ -427,21 +447,22 @@ impl Ledger { /// Snapshot the single-row `archive_state` table as a JSON object — /// `{ schema_version, upstream_cursors_json, last_built_at, - /// last_rebuild_at }`. Powers `state_status`'s `archive` block; kept + /// last_rebuild_at, last_write_at_ms }`. Powers `state_status`'s `archive` block; kept /// here rather than at the SDK verb so callers don't have to bind /// to rusqlite directly to read first-party rows. pub fn read_archive_state_json(&self) -> Result { - let row: (i64, String, Option, Option) = self.conns.burn.query_row( - "SELECT schema_version, upstream_cursors_json, last_built_at, last_rebuild_at \ + let row: (i64, String, Option, Option, Option) = self.conns.burn.query_row( + "SELECT schema_version, upstream_cursors_json, last_built_at, last_rebuild_at, last_write_at_ms \ FROM archive_state WHERE id = 1", [], - |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)), + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?)), )?; let value = serde_json::json!({ "schema_version": row.0, "upstream_cursors_json": row.1, "last_built_at": row.2, "last_rebuild_at": row.3, + "last_write_at_ms": row.4, }); Ok(value.to_string()) } @@ -564,7 +585,8 @@ impl Ledger { SET upstream_cursors_json = '{}', \ last_built_at = NULL, \ last_rebuild_at = NULL, \ - source_fingerprint = '' \ + source_fingerprint = '', \ + last_write_at_ms = NULL \ WHERE id = 1", [], )?; diff --git a/crates/relayburn-sdk/src/ledger/config.rs b/crates/relayburn-sdk/src/ledger/config.rs index 2b1fbd3c..c905065f 100644 --- a/crates/relayburn-sdk/src/ledger/config.rs +++ b/crates/relayburn-sdk/src/ledger/config.rs @@ -2,8 +2,8 @@ //! //! Mirrors `packages/ledger/src/config.ts`: a small JSON file at //! `$RELAYBURN_HOME/config.json` with environment-variable overrides for -//! the two knobs ingest cares about (`content.store` and -//! `content.retentionDays`). The TS source-of-truth co-locates this with +//! content storage/retention and the read staleness threshold. The +//! TS source-of-truth co-locates this with //! `@relayburn/ledger`, so the Rust port keeps the same home — the //! ledger crate already depends on `relayburn-reader` for //! [`ContentStoreMode`], and ingest (#277, #278) already depends on the @@ -23,9 +23,14 @@ use crate::reader::ContentStoreMode; use crate::ledger::error::Result; use crate::ledger::paths::ledger_home; +#[cfg(test)] +pub(crate) static CONFIG_ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + /// Default content retention window in days. Matches TS /// `DEFAULT_RETENTION_DAYS`. pub const DEFAULT_RETENTION_DAYS: f64 = 90.0; +/// Default age after which read results are considered stale. +pub const DEFAULT_STALE_AFTER_HOURS: f64 = 24.0; /// Retention window for content rows. Mirrors the TS /// `number | 'forever'` shape; `Forever` disables TTL-based pruning. @@ -67,6 +72,20 @@ pub struct BurnConfig { pub content: ContentConfig, } +/// Controls when read/report surfaces flag the ledger as stale. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct StalenessConfig { + pub threshold_hours: f64, +} + +impl Default for StalenessConfig { + fn default() -> Self { + Self { + threshold_hours: DEFAULT_STALE_AFTER_HOURS, + } + } +} + impl Default for BurnConfig { /// Defaults match TS `DEFAULT_CONFIG`: /// `{ content: { store: 'full', retentionDays: 90 } }`. @@ -95,9 +114,10 @@ pub fn config_path_at_home(home: &Path) -> PathBuf { home.join("config.json") } -/// Load the user config: read the JSON file (if present), then layer the -/// `RELAYBURN_CONTENT_STORE` and `RELAYBURN_CONTENT_TTL_DAYS` env vars on -/// top, falling back to [`BurnConfig::default`]. +/// Load the content config: read the JSON file (if present), then layer the +/// `RELAYBURN_CONTENT_STORE` and `RELAYBURN_CONTENT_TTL_DAYS` env vars on top, +/// falling back to [`BurnConfig::default`]. Use [`load_staleness_config`] for +/// the separately version-compatible read/report threshold. /// /// Mirrors the TS `loadConfig()` precedence: env overrides file overrides /// default. A missing file is the common case and not an error; malformed @@ -120,6 +140,39 @@ pub fn load_config_with_home(home: Option<&Path>) -> Result { } } +/// Load the read/report staleness settings from the default config path. +/// +/// This is separate from [`BurnConfig`] so adding the optional staleness +/// feature does not break embedders that construct that public type with a +/// struct literal. +pub fn load_staleness_config() -> Result { + load_staleness_config_at(&config_path()) +} + +/// Like [`load_staleness_config`], but resolves the config under an explicit +/// ledger home when supplied. +pub fn load_staleness_config_with_home(home: Option<&Path>) -> Result { + match home { + Some(h) => load_staleness_config_at(&config_path_at_home(h)), + None => load_staleness_config(), + } +} + +/// Load staleness settings from an explicit config path. +pub fn load_staleness_config_at(path: &Path) -> Result { + let from_file = read_config_file(path); + Ok(StalenessConfig { + threshold_hours: pick_staleness_threshold( + std::env::var("RELAYBURN_STALE_AFTER_HOURS").ok().as_deref(), + from_file + .as_ref() + .and_then(|c| c.staleness.as_ref()) + .and_then(|c| c.threshold_hours.as_ref()), + DEFAULT_STALE_AFTER_HOURS, + ), + }) +} + /// Load with an explicit config path. Tests use this to avoid touching /// `$HOME/.agentworkforce/burn/config.json`. pub fn load_config_at(path: &Path) -> Result { @@ -156,6 +209,14 @@ pub fn load_config_at(path: &Path) -> Result { struct RawConfig { #[serde(default)] content: Option, + #[serde(default)] + staleness: Option, +} + +#[derive(Debug, Default, Deserialize, Serialize)] +struct RawStaleness { + #[serde(default, rename = "thresholdHours")] + threshold_hours: Option, } #[derive(Debug, Default, Deserialize, Serialize)] @@ -286,24 +347,37 @@ fn normalize_retention_f64(f: f64) -> Option { Some(Retention::Days(f)) } +fn pick_staleness_threshold( + env: Option<&str>, + file: Option<&serde_json::Value>, + default: f64, +) -> f64 { + env.and_then(|s| s.trim().parse::().ok()) + .filter(|v| v.is_finite()) + .or_else(|| { + file.and_then(serde_json::Value::as_f64) + .filter(|v| v.is_finite()) + }) + .unwrap_or(default) +} + #[cfg(test)] mod tests { use super::*; - use std::sync::Mutex; use tempfile::TempDir; // The picker functions read process-wide env vars. Serialize tests // that touch them so a parallel test run doesn't see a leaked value // from a peer. - static ENV_LOCK: Mutex<()> = Mutex::new(()); - fn with_clean_env(f: F) { - let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let _g = CONFIG_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); std::env::remove_var("RELAYBURN_CONTENT_STORE"); std::env::remove_var("RELAYBURN_CONTENT_TTL_DAYS"); + std::env::remove_var("RELAYBURN_STALE_AFTER_HOURS"); f(); std::env::remove_var("RELAYBURN_CONTENT_STORE"); std::env::remove_var("RELAYBURN_CONTENT_TTL_DAYS"); + std::env::remove_var("RELAYBURN_STALE_AFTER_HOURS"); } #[test] @@ -313,10 +387,27 @@ mod tests { let cfg = load_config_at(&tmp.path().join("config.json")).unwrap(); assert_eq!(cfg.content.store, ContentStoreMode::Full); assert_eq!(cfg.content.retention_days, Retention::Days(90.0)); + assert_eq!( + load_staleness_config_at(&tmp.path().join("config.json")) + .unwrap() + .threshold_hours, + 24.0 + ); assert_eq!(cfg, BurnConfig::default()); }); } + #[test] + fn burn_config_preserves_the_public_struct_literal_shape() { + let cfg = BurnConfig { + content: ContentConfig { + store: ContentStoreMode::Full, + retention_days: Retention::Days(90.0), + }, + }; + assert_eq!(cfg, BurnConfig::default()); + } + #[test] fn file_overrides_default() { with_clean_env(|| { @@ -377,6 +468,37 @@ mod tests { }); } + #[test] + fn staleness_threshold_file_and_env_overrides() { + with_clean_env(|| { + let tmp = TempDir::new().unwrap(); + let path = tmp.path().join("config.json"); + std::fs::write(&path, r#"{"staleness":{"thresholdHours":6}}"#).unwrap(); + assert_eq!( + load_staleness_config_at(&path).unwrap().threshold_hours, + 6.0 + ); + std::env::set_var("RELAYBURN_STALE_AFTER_HOURS", "2.5"); + assert_eq!( + load_staleness_config_at(&path).unwrap().threshold_hours, + 2.5 + ); + }); + } + + #[test] + fn any_negative_disables_staleness_warning() { + with_clean_env(|| { + let tmp = TempDir::new().unwrap(); + let path = tmp.path().join("config.json"); + std::fs::write(&path, r#"{"staleness":{"thresholdHours":-10}}"#).unwrap(); + assert_eq!( + load_staleness_config_at(&path).unwrap().threshold_hours, + -10.0 + ); + }); + } + #[test] fn empty_env_string_does_not_zero_retention() { with_clean_env(|| { diff --git a/crates/relayburn-sdk/src/ledger/db.rs b/crates/relayburn-sdk/src/ledger/db.rs index b1c24bb5..cb4d3e0c 100644 --- a/crates/relayburn-sdk/src/ledger/db.rs +++ b/crates/relayburn-sdk/src/ledger/db.rs @@ -241,6 +241,43 @@ fn migrate_burn_schema(conn: &Connection) -> Result<()> { )?; } + if current_version < 7 { + match conn.execute( + "ALTER TABLE archive_state ADD COLUMN last_write_at_ms INTEGER", + [], + ) { + Ok(_) => {} + Err(rusqlite::Error::SqliteFailure(_, Some(msg))) + if msg.contains("duplicate column name") => {} + Err(e) => return Err(e.into()), + } + // Legacy ledgers have no write clock. Seed conservatively from the + // newest event timestamp so an old snapshot is warned about on its + // first post-upgrade read instead of being made artificially fresh by + // the schema migration itself. + conn.execute( + "UPDATE archive_state + SET last_write_at_ms = ( + SELECT MAX(ms) FROM ( + SELECT CAST(strftime('%s', MAX(ts)) AS INTEGER) * 1000 + + CAST(substr(strftime('%f', MAX(ts)), 4, 3) AS INTEGER) AS ms FROM turns + UNION ALL SELECT CAST(strftime('%s', MAX(ts)) AS INTEGER) * 1000 + + CAST(substr(strftime('%f', MAX(ts)), 4, 3) AS INTEGER) FROM compactions + UNION ALL SELECT CAST(strftime('%s', MAX(ts)) AS INTEGER) * 1000 + + CAST(substr(strftime('%f', MAX(ts)), 4, 3) AS INTEGER) FROM relationships + UNION ALL SELECT CAST(strftime('%s', MAX(ts)) AS INTEGER) * 1000 + + CAST(substr(strftime('%f', MAX(ts)), 4, 3) AS INTEGER) FROM tool_result_events + UNION ALL SELECT CAST(strftime('%s', MAX(ts)) AS INTEGER) * 1000 + + CAST(substr(strftime('%f', MAX(ts)), 4, 3) AS INTEGER) FROM user_turns + UNION ALL SELECT CAST(strftime('%s', MAX(end_ts)) AS INTEGER) * 1000 + + CAST(substr(strftime('%f', MAX(end_ts)), 4, 3) AS INTEGER) FROM inferences + ) + ), schema_version = 7 + WHERE id = 1", + [], + )?; + } + // The `idx_turns_stop_reason` index is created here rather than in // the static DDL so a legacy v1 table (no `stop_reason` column yet) // doesn't fail the DDL pre-pass. By this point the column either diff --git a/crates/relayburn-sdk/src/ledger/schema.rs b/crates/relayburn-sdk/src/ledger/schema.rs index 488def6c..aa68d6c8 100644 --- a/crates/relayburn-sdk/src/ledger/schema.rs +++ b/crates/relayburn-sdk/src/ledger/schema.rs @@ -65,7 +65,10 @@ pub const DERIVABLE_TABLES: &[&str] = &[ /// per-file deserialize. Blanked by `state rebuild` / `state reset` so the /// next ingest does not trust source state captured before derived rows were /// wiped. (#468) -pub const SCHEMA_VERSION: u32 = 6; +/// - `7`: adds `archive_state.last_write_at_ms INTEGER`, updated once per +/// successful derived-ledger write batch, so read surfaces can distinguish +/// a current ledger from one that has not received data recently. (#507) +pub const SCHEMA_VERSION: u32 = 7; /// DDL for `burn.sqlite`. Idempotent (`IF NOT EXISTS`) so re-applying on /// startup is a no-op once the tables exist. @@ -232,11 +235,12 @@ CREATE TABLE IF NOT EXISTS archive_state ( last_rebuild_at TEXT, -- Cheap source-side change gate: `count:totalBytes:hash` over the -- session files ingest scans. Empty until the first ingest records it. - source_fingerprint TEXT NOT NULL DEFAULT '' + source_fingerprint TEXT NOT NULL DEFAULT '', + last_write_at_ms INTEGER ); INSERT INTO archive_state (id, schema_version) - VALUES (1, 6) + VALUES (1, 7) ON CONFLICT(id) DO NOTHING; "#; diff --git a/crates/relayburn-sdk/src/ledger/tests.rs b/crates/relayburn-sdk/src/ledger/tests.rs index b5b162f9..73a7f248 100644 --- a/crates/relayburn-sdk/src/ledger/tests.rs +++ b/crates/relayburn-sdk/src/ledger/tests.rs @@ -531,6 +531,31 @@ fn stamp_synthesizes_spawn_env_relationship() { assert_eq!(rels[0].relationship_type, RelationshipType::Subagent); assert_eq!(rels[0].related_session_id.as_deref(), Some("parent-1")); assert_eq!(rels[0].agent_id.as_deref(), Some("child-1")); + assert!( + l.last_write_at_ms().unwrap().is_some(), + "a newly synthesized derived relationship should refresh ledger freshness" + ); +} + +#[test] +fn annotation_only_stamp_does_not_refresh_ledger_freshness() { + let tmp = TempDir::new().unwrap(); + let mut l = open_in(&tmp); + let mut enrichment = BTreeMap::new(); + enrichment.insert("role".into(), "fix-bug".into()); + let stamp = Stamp::new( + "2025-01-01T00:00:00Z", + StampSelector { + session_id: Some("s1".into()), + ..Default::default() + }, + enrichment, + ) + .unwrap(); + + l.append_stamp(&stamp).unwrap(); + + assert_eq!(l.last_write_at_ms().unwrap(), None); } #[test] @@ -739,7 +764,7 @@ fn invalid_session_id_in_content_rejected() { /// column on `turns`, `archive_state.schema_version = 1`) opens cleanly /// against the 3.0 SDK, the column is back-added by the in-place /// migration, and the stored version bumps forward to the current -/// `SCHEMA_VERSION` (6 after #436 + #435 + #434 + #468 chained on top of +/// `SCHEMA_VERSION` (7 after #436 + #435 + #434 + #468 + #507 chained on top of /// #437). /// Existing rows stay `NULL` for every back-added column until rewritten. #[test] @@ -776,8 +801,8 @@ fn legacy_v1_ledger_migrates_to_v2_on_open_and_adds_stop_reason_column() { INSERT INTO turns (source, session_id, message_id, ts, project, project_key, record_json, content_fingerprint) VALUES ('claude-code', 'legacy-sess', 'legacy-msg', - '2025-01-01T00:00:00Z', NULL, NULL, - '{\"v\":1,\"source\":\"claude-code\",\"sessionId\":\"legacy-sess\",\"messageId\":\"legacy-msg\",\"turnIndex\":0,\"ts\":\"2025-01-01T00:00:00Z\",\"model\":\"claude-sonnet-4-6\",\"usage\":{\"input\":0,\"output\":0,\"reasoning\":0,\"cacheRead\":0,\"cacheCreate5m\":0,\"cacheCreate1h\":0},\"toolCalls\":[]}', + '2025-01-01T00:00:00.123Z', NULL, NULL, + '{\"v\":1,\"source\":\"claude-code\",\"sessionId\":\"legacy-sess\",\"messageId\":\"legacy-msg\",\"turnIndex\":0,\"ts\":\"2025-01-01T00:00:00.123Z\",\"model\":\"claude-sonnet-4-6\",\"usage\":{\"input\":0,\"output\":0,\"reasoning\":0,\"cacheRead\":0,\"cacheCreate5m\":0,\"cacheCreate1h\":0},\"toolCalls\":[]}', 'legacy-fp'); ", ) @@ -787,7 +812,7 @@ fn legacy_v1_ledger_migrates_to_v2_on_open_and_adds_stop_reason_column() { // Step 2: open through the SDK. The migration must: // a) add `turns.stop_reason TEXT`, // b) bump archive_state.schema_version forward to the current - // `SCHEMA_VERSION` (chained v1 → v2 → v3 → v4 → v5 → v6), + // `SCHEMA_VERSION` (chained v1 → v2 → v3 → v4 → v5 → v6 → v7), // c) leave the legacy row's stop_reason as NULL. let l = Ledger::open(&layout.burn, &layout.content).unwrap(); let version: i64 = l @@ -799,11 +824,11 @@ fn legacy_v1_ledger_migrates_to_v2_on_open_and_adds_stop_reason_column() { |r| r.get(0), ) .unwrap(); - // Current `SCHEMA_VERSION` is 6 (chained #437 v2 + #436 v3 + #435 - // v4 + #434 v5 + #468 v6); the migration must walk every step in one + // Current `SCHEMA_VERSION` is 7 (chained #437 v2 + #436 v3 + #435 + // v4 + #434 v5 + #468 v6 + #507 v7); the migration must walk every step in one // open() call. assert_eq!( - version, 6, + version, 7, "open must bump v1 forward to the current schema version" ); @@ -822,6 +847,15 @@ fn legacy_v1_ledger_migrates_to_v2_on_open_and_adds_stop_reason_column() { archive_cols.iter().any(|c| c == "source_fingerprint"), "v6 migration must add archive_state.source_fingerprint" ); + assert!( + archive_cols.iter().any(|c| c == "last_write_at_ms"), + "v7 migration must add archive_state.last_write_at_ms" + ); + assert_eq!( + l.last_write_at_ms().unwrap(), + Some(1_735_689_600_123), + "v7 migration must preserve event timestamp milliseconds" + ); let column_names: Vec = l .conns @@ -884,6 +918,36 @@ fn legacy_v1_ledger_migrates_to_v2_on_open_and_adds_stop_reason_column() { let _ = Ledger::open(&layout.burn, &layout.content).unwrap(); } +#[test] +fn v7_migration_seeds_inference_only_activity() { + let tmp = TempDir::new().unwrap(); + let layout = LedgerLayout::under(tmp.path()); + drop(Ledger::open(&layout.burn, &layout.content).unwrap()); + + { + let conn = rusqlite::Connection::open(&layout.burn).unwrap(); + conn.execute( + "INSERT INTO inferences + (source, session_id, request_id, request_id_source, turn_id, + model, kind, start_ts, end_ts, record_json) + VALUES ('codex', 's1', 'r1', 'explicit', 't1', 'gpt-5', 'text', + '2025-01-01T00:00:00.100Z', '2025-01-01T00:00:00.987Z', '{}')", + [], + ) + .unwrap(); + conn.execute( + "UPDATE archive_state + SET schema_version = 6, last_write_at_ms = NULL + WHERE id = 1", + [], + ) + .unwrap(); + } + + let ledger = Ledger::open(&layout.burn, &layout.content).unwrap(); + assert_eq!(ledger.last_write_at_ms().unwrap(), Some(1_735_689_600_987)); +} + #[test] fn schema_too_new_is_rejected() { // Defensive: if a future build wrote a higher schema_version, this diff --git a/crates/relayburn-sdk/src/ledger/writer.rs b/crates/relayburn-sdk/src/ledger/writer.rs index bd12f577..6f05d465 100644 --- a/crates/relayburn-sdk/src/ledger/writer.rs +++ b/crates/relayburn-sdk/src/ledger/writer.rs @@ -41,6 +41,18 @@ fn now_lex_token() -> String { format!("ts:{:020}.{:09}", secs, nanos_part) } +fn touch_last_write(tx: &rusqlite::Transaction<'_>) -> Result<()> { + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64; + tx.execute( + "UPDATE archive_state SET last_write_at_ms = ? WHERE id = 1", + params![now_ms], + )?; + Ok(()) +} + pub(crate) fn append_turns(conn: &mut Connection, turns: &[TurnRecord]) -> Result { if turns.is_empty() { return Ok(0); @@ -93,6 +105,9 @@ pub(crate) fn append_turns(conn: &mut Connection, turns: &[TurnRecord]) -> Resul } } } + if appended > 0 { + touch_last_write(&tx)?; + } tx.commit()?; Ok(appended) } @@ -122,6 +137,9 @@ pub(crate) fn append_compactions( } } } + if appended > 0 { + touch_last_write(&tx)?; + } tx.commit()?; Ok(appended) } @@ -159,6 +177,9 @@ pub(crate) fn append_relationships( } } } + if appended > 0 { + touch_last_write(&tx)?; + } tx.commit()?; Ok(appended) } @@ -199,6 +220,9 @@ pub(crate) fn append_tool_result_events( } } } + if appended > 0 { + touch_last_write(&tx)?; + } tx.commit()?; Ok(appended) } @@ -242,6 +266,9 @@ pub(crate) fn append_inferences(conn: &mut Connection, records: &[Inference]) -> } } } + if appended > 0 { + touch_last_write(&tx)?; + } tx.commit()?; Ok(appended) } @@ -277,6 +304,9 @@ pub(crate) fn append_user_turns( } } } + if appended > 0 { + touch_last_write(&tx)?; + } tx.commit()?; Ok(appended) } @@ -292,6 +322,7 @@ pub(crate) fn append_stamp(conn: &mut Connection, stamp: &Stamp) -> Result<()> { let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; let written_at = now_lex_token(); + let mut derived_rows_written = false; { tx.prepare( "INSERT INTO stamps (source, session_id, ts, selector_json, enrichment_json, written_at) @@ -308,23 +339,28 @@ pub(crate) fn append_stamp(conn: &mut Connection, stamp: &Stamp) -> Result<()> { if let Some(rel) = synthesized { let id = relationship_id_fingerprint(&rel); let json = serde_json::to_string(&rel)?; - tx.prepare( - "INSERT OR IGNORE INTO relationships + let changed = tx + .prepare( + "INSERT OR IGNORE INTO relationships (id_fingerprint, source, session_id, related_session_id, relationship_type, ts, record_json) VALUES (?, ?, ?, ?, ?, ?, ?)", - )? - .execute(params![ - id, - rel.source.wire_str(), - rel.session_id, - rel.related_session_id, - rel.relationship_type.wire_str(), - rel.ts, - json, - ])?; + )? + .execute(params![ + id, + rel.source.wire_str(), + rel.session_id, + rel.related_session_id, + rel.relationship_type.wire_str(), + rel.ts, + json, + ])?; + derived_rows_written = changed > 0; } } + if derived_rows_written { + touch_last_write(&tx)?; + } tx.commit()?; Ok(()) } diff --git a/crates/relayburn-sdk/src/lib.rs b/crates/relayburn-sdk/src/lib.rs index 445b8ea1..cc2285c9 100644 --- a/crates/relayburn-sdk/src/lib.rs +++ b/crates/relayburn-sdk/src/lib.rs @@ -34,6 +34,9 @@ //! against multiple ledgers in the same process. use std::path::PathBuf; +use std::time::{SystemTime, UNIX_EPOCH}; + +use serde::{Deserialize, Serialize}; // Internal lower-stack modules. Order matches the dependency graph // (reader -> ledger -> analyze -> ingest); the verb modules below pull from @@ -81,10 +84,12 @@ pub use crate::reader::{ pub use crate::ledger::{ burn_sqlite_path, config_path, config_path_at_home, content_sqlite_path, is_valid_session_id, - ledger_home, load_config, load_config_with_home, BurnConfig, ContentConfig, EnrichedTurn, - Enrichment, Ledger as RawLedger, LedgerError, LedgerFingerprintScope, MessageRange, PruneStats, - Query, RebuildSummary, ResetSummary, Retention, SearchHit, SearchOptions, Stamp, StampError, - StampSelector, DEFAULT_RETENTION_DAYS, + ledger_home, load_config, load_config_with_home, load_staleness_config, + load_staleness_config_at, load_staleness_config_with_home, BurnConfig, ContentConfig, + EnrichedTurn, Enrichment, Ledger as RawLedger, LedgerError, LedgerFingerprintScope, + MessageRange, PruneStats, Query, RebuildSummary, ResetSummary, Retention, SearchHit, + SearchOptions, StalenessConfig, Stamp, StampError, StampSelector, DEFAULT_RETENTION_DAYS, + DEFAULT_STALE_AFTER_HOURS, }; pub use crate::analyze::{ @@ -192,6 +197,7 @@ impl LedgerOpenOptions { /// sharing one through a lock. pub struct LedgerHandle { pub(crate) inner: RawLedger, + config_home: PathBuf, } impl LedgerHandle { @@ -205,6 +211,50 @@ impl LedgerHandle { pub fn raw_mut(&mut self) -> &mut RawLedger { &mut self.inner } + + /// Return freshness metadata for read/report consumers. The SDK exposes + /// this as data; presenters decide whether and how to warn. + pub fn ledger_freshness(&self) -> anyhow::Result { + let now_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64; + self.ledger_freshness_at(now_ms) + } + + /// Deterministic variant of [`Self::ledger_freshness`] for callers that + /// already own a clock and for threshold-boundary tests. + pub fn ledger_freshness_at(&self, now_ms: u64) -> anyhow::Result { + let config = load_staleness_config_with_home(Some(&self.config_home))?; + let disabled = config.threshold_hours < 0.0; + let stale_after_ms = if disabled { + None + } else { + Some((config.threshold_hours * 3_600_000.0) as u64) + }; + let last_write_at_ms = self.inner.last_write_at_ms()?; + let stale = !disabled + && last_write_at_ms + .map(|last| now_ms.saturating_sub(last) > stale_after_ms.unwrap_or_default()) + .unwrap_or(true); + Ok(LedgerFreshness { + last_write_at_ms, + stale_after_ms, + stale, + }) + } +} + +/// Shared staleness flag returned to SDK, Node, and MCP consumers. Its clock +/// tracks event/derived-row writes in `burn.sqlite`, not content-sidecar-only +/// persistence in `content.sqlite`. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct LedgerFreshness { + pub last_write_at_ms: Option, + /// Configured threshold, or `None` when staleness warnings are disabled. + pub stale_after_ms: Option, + pub stale: bool, } /// Namespace type for the open verb. Matches the TS surface @@ -216,9 +266,159 @@ impl Ledger { /// Open the ledger described by `opts`, applying schema DDL if needed, /// and return a [`LedgerHandle`] for the verbs in this crate. pub fn open(opts: LedgerOpenOptions) -> anyhow::Result { + let config_home = opts.home.clone().unwrap_or_else(ledger_home); let burn = opts.resolve_burn_path(); let content = opts.resolve_content_path(); let inner = RawLedger::open(&burn, &content)?; - Ok(LedgerHandle { inner }) + Ok(LedgerHandle { inner, config_home }) + } +} + +#[cfg(test)] +mod freshness_tests { + use super::*; + use rusqlite::{params, Connection}; + use tempfile::TempDir; + + struct CleanStaleEnv { + _lock: std::sync::MutexGuard<'static, ()>, + previous: Option, + } + + impl CleanStaleEnv { + fn new() -> Self { + let lock = crate::ledger::CONFIG_ENV_LOCK + .lock() + .unwrap_or_else(|e| e.into_inner()); + let previous = std::env::var_os("RELAYBURN_STALE_AFTER_HOURS"); + std::env::remove_var("RELAYBURN_STALE_AFTER_HOURS"); + Self { + _lock: lock, + previous, + } + } + } + + impl Drop for CleanStaleEnv { + fn drop(&mut self) { + match self.previous.take() { + Some(value) => std::env::set_var("RELAYBURN_STALE_AFTER_HOURS", value), + None => std::env::remove_var("RELAYBURN_STALE_AFTER_HOURS"), + } + } + } + + fn set_last_write(home: &std::path::Path, value: u64) { + let conn = Connection::open(home.join("burn.sqlite")).unwrap(); + conn.execute( + "UPDATE archive_state SET last_write_at_ms = ? WHERE id = 1", + params![value as i64], + ) + .unwrap(); + } + + #[test] + fn freshness_is_strictly_past_threshold() { + let _env = CleanStaleEnv::new(); + let tmp = TempDir::new().unwrap(); + let handle = Ledger::open(LedgerOpenOptions::with_home(tmp.path())).unwrap(); + set_last_write(tmp.path(), 1_000); + let threshold = 24 * 60 * 60 * 1_000; + + let boundary = handle.ledger_freshness_at(1_000 + threshold).unwrap(); + assert!(!boundary.stale, "equal to the threshold is still fresh"); + assert_eq!(boundary.last_write_at_ms, Some(1_000)); + assert!( + handle + .ledger_freshness_at(1_000 + threshold + 1) + .unwrap() + .stale + ); + } + + #[test] + fn config_override_changes_stale_decision() { + let _env = CleanStaleEnv::new(); + let tmp = TempDir::new().unwrap(); + std::fs::write( + tmp.path().join("config.json"), + r#"{"staleness":{"thresholdHours":2}}"#, + ) + .unwrap(); + let handle = Ledger::open(LedgerOpenOptions::with_home(tmp.path())).unwrap(); + set_last_write(tmp.path(), 1_000); + + let status = handle + .ledger_freshness_at(1_000 + 2 * 3_600_000 + 1) + .unwrap(); + assert_eq!(status.stale_after_ms, Some(7_200_000)); + assert!(status.stale); + } + + #[test] + fn real_ledger_write_batch_updates_clock_and_is_fresh() { + let _env = CleanStaleEnv::new(); + let tmp = TempDir::new().unwrap(); + let mut handle = Ledger::open(LedgerOpenOptions::with_home(tmp.path())).unwrap(); + let turn: TurnRecord = serde_json::from_value(serde_json::json!({ + "v": 1, + "source": "codex", + "sessionId": "s", + "messageId": "m", + "turnIndex": 0, + "ts": "2026-01-01T00:00:00.000Z", + "model": "gpt-5.2-codex", + "usage": {"input": 1, "output": 1, "reasoning": 0, "cacheRead": 0, "cacheCreate5m": 0, "cacheCreate1h": 0}, + "toolCalls": [] + })) + .unwrap(); + handle.raw_mut().append_turns(&[turn]).unwrap(); + + let now_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + let status = handle.ledger_freshness_at(now_ms).unwrap(); + assert!(status.last_write_at_ms.is_some()); + assert!(!status.stale); + } + + #[test] + fn never_written_is_stale_but_future_clock_is_fresh() { + let _env = CleanStaleEnv::new(); + let tmp = TempDir::new().unwrap(); + let handle = Ledger::open(LedgerOpenOptions::with_home(tmp.path())).unwrap(); + assert!(handle.ledger_freshness_at(1_000).unwrap().stale); + + set_last_write(tmp.path(), 2_000); + let future = handle.ledger_freshness_at(1_000).unwrap(); + assert!(!future.stale); + } + + #[test] + fn disabled_threshold_suppresses_even_never_written() { + let _env = CleanStaleEnv::new(); + let tmp = TempDir::new().unwrap(); + std::fs::write( + tmp.path().join("config.json"), + r#"{"staleness":{"thresholdHours":-10}}"#, + ) + .unwrap(); + let handle = Ledger::open(LedgerOpenOptions::with_home(tmp.path())).unwrap(); + let status = handle.ledger_freshness_at(1_000).unwrap(); + assert!(!status.stale); + assert_eq!(status.stale_after_ms, None); + } + + #[test] + fn reset_clears_write_clock_and_reports_stale() { + let _env = CleanStaleEnv::new(); + let tmp = TempDir::new().unwrap(); + let mut handle = Ledger::open(LedgerOpenOptions::with_home(tmp.path())).unwrap(); + set_last_write(tmp.path(), 1_000); + handle.raw_mut().reset().unwrap(); + let status = handle.ledger_freshness_at(2_000).unwrap(); + assert_eq!(status.last_write_at_ms, None); + assert!(status.stale); } } diff --git a/crates/relayburn-sdk/src/query_verbs/state.rs b/crates/relayburn-sdk/src/query_verbs/state.rs index 6b1ff184..fb049f31 100644 --- a/crates/relayburn-sdk/src/query_verbs/state.rs +++ b/crates/relayburn-sdk/src/query_verbs/state.rs @@ -52,6 +52,8 @@ pub struct ArchiveStateStatus { pub last_built_at: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub last_rebuild_at: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_write_at_ms: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -179,12 +181,15 @@ fn read_archive_state(ledger: &crate::RawLedger) -> Result { last_built_at: Option, #[serde(default)] last_rebuild_at: Option, + #[serde(default)] + last_write_at_ms: Option, } let raw: Raw = serde_json::from_str(&json).map_err(|e| anyhow::anyhow!(e))?; Ok(ArchiveStateStatus { schema_version: raw.schema_version, last_built_at: raw.last_built_at, last_rebuild_at: raw.last_rebuild_at, + last_write_at_ms: raw.last_write_at_ms, }) } diff --git a/crates/relayburn-sdk/src/query_verbs/tests.rs b/crates/relayburn-sdk/src/query_verbs/tests.rs index 5f345ae6..5f4f85fa 100644 --- a/crates/relayburn-sdk/src/query_verbs/tests.rs +++ b/crates/relayburn-sdk/src/query_verbs/tests.rs @@ -1607,7 +1607,8 @@ fn state_status_reports_zero_rows_on_fresh_ledger() { // (#434 `inferences`), v4 (#435 `turns.subagent_id`), v3 (#436 // `tool_result_events.output_bytes` / `output_truncated`) and v2 // (#437 `turns.stop_reason`). - assert_eq!(s.archive.schema_version, 6); + assert_eq!(s.archive.schema_version, 7); + assert!(s.archive.last_write_at_ms.is_none()); assert!(s.archive.last_built_at.is_none()); assert!(s.archive.last_rebuild_at.is_none()); } diff --git a/packages/mcp/CHANGELOG.md b/packages/mcp/CHANGELOG.md index 6a17e22e..50446fa2 100644 --- a/packages/mcp/CHANGELOG.md +++ b/packages/mcp/CHANGELOG.md @@ -4,6 +4,7 @@ All notable changes to `@relayburn/mcp`. ## [Unreleased] +- Read-tool responses now include `ledgerFreshness` with the ledger's last-write timestamp, threshold, and stale flag. - Cost output recognizes Claude 5 and GPT-5.6 models, prefers first-party tariffs, and applies long-context price tiers. ## [4.0.0] - 2026-06-23 diff --git a/packages/mcp/src/end-to-end.test.ts b/packages/mcp/src/end-to-end.test.ts index fc3bde16..ab51dc1e 100644 --- a/packages/mcp/src/end-to-end.test.ts +++ b/packages/mcp/src/end-to-end.test.ts @@ -40,6 +40,11 @@ describe('end-to-end: spawn server, call burn__sessionCost, verify cost', () => const output = new PassThrough(); const responses = collectResponses(output); const sessionCost = createSessionCostTool({ + ledgerFreshness: async () => ({ + lastWriteAtMs: 1, + staleAfterMs: 86_400_000, + stale: false, + }), defaultSessionId: 'S', sessionCost: async (opts) => ({ sessionId: opts.session ?? null, diff --git a/packages/mcp/src/tools/fingerprint.test.ts b/packages/mcp/src/tools/fingerprint.test.ts index f47af80d..2df26570 100644 --- a/packages/mcp/src/tools/fingerprint.test.ts +++ b/packages/mcp/src/tools/fingerprint.test.ts @@ -3,18 +3,38 @@ import { describe, it } from 'node:test'; import { createFingerprintTool, type FingerprintResult } from './fingerprint.js'; +const FRESHNESS_DEP = { + ledgerFreshness: async () => ({ lastWriteAtMs: 1, staleAfterMs: 86_400_000, stale: false }), +}; + describe('createFingerprintTool', () => { it('returns the SDK fingerprint string verbatim', async () => { const tool = createFingerprintTool({ + ...FRESHNESS_DEP, fingerprint: async () => ({ fingerprint: '42:1700000000:9876' }), }); const result = (await tool.handler({})) as FingerprintResult; assert.equal(result.fingerprint, '42:1700000000:9876'); + assert.deepEqual(result.ledgerFreshness, { + lastWriteAtMs: 1, + staleAfterMs: 86_400_000, + stale: false, + }); + }); + + it('returns stale freshness metadata unchanged', async () => { + const tool = createFingerprintTool({ + ledgerFreshness: async () => ({ lastWriteAtMs: 1, staleAfterMs: 2, stale: true }), + fingerprint: async () => ({ fingerprint: '1:1:1' }), + }); + const result = (await tool.handler({})) as FingerprintResult; + assert.deepEqual(result.ledgerFreshness, { lastWriteAtMs: 1, staleAfterMs: 2, stale: true }); }); it('passes sessionId through as session', async () => { let captured: { session?: string; project?: string } = {}; const tool = createFingerprintTool({ + ...FRESHNESS_DEP, fingerprint: async (opts) => { captured = opts; return { fingerprint: '1:1:1' }; @@ -28,6 +48,7 @@ describe('createFingerprintTool', () => { it('passes project through', async () => { let captured: { session?: string; project?: string } = {}; const tool = createFingerprintTool({ + ...FRESHNESS_DEP, fingerprint: async (opts) => { captured = opts; return { fingerprint: '1:1:1' }; @@ -40,6 +61,7 @@ describe('createFingerprintTool', () => { it('rejects sessionId + project together', async () => { const tool = createFingerprintTool({ + ...FRESHNESS_DEP, fingerprint: async () => ({ fingerprint: 'unreachable' }), }); await assert.rejects( diff --git a/packages/mcp/src/tools/fingerprint.ts b/packages/mcp/src/tools/fingerprint.ts index 5766e691..9e3e2a20 100644 --- a/packages/mcp/src/tools/fingerprint.ts +++ b/packages/mcp/src/tools/fingerprint.ts @@ -1,5 +1,5 @@ -import { fingerprint as sdkFingerprint } from '@relayburn/sdk'; -import type { FingerprintResult as SdkFingerprintResult } from '@relayburn/sdk'; +import { fingerprint as sdkFingerprint, ledgerFreshness as sdkLedgerFreshness } from '@relayburn/sdk'; +import type { FingerprintResult as SdkFingerprintResult, LedgerFreshness } from '@relayburn/sdk'; import type { ToolDefinition } from '../types.js'; @@ -8,7 +8,7 @@ export interface FingerprintInput { project?: string; } -export type FingerprintResult = SdkFingerprintResult; +export type FingerprintResult = SdkFingerprintResult & { ledgerFreshness: LedgerFreshness }; export interface FingerprintDeps { /** @@ -19,6 +19,7 @@ export interface FingerprintDeps { session?: string; project?: string; }) => Promise; + ledgerFreshness?: () => Promise; } /** @@ -29,13 +30,16 @@ export interface FingerprintDeps { */ export function createFingerprintTool(deps: FingerprintDeps = {}): ToolDefinition { const callFingerprint = deps.fingerprint ?? sdkFingerprint; + const callLedgerFreshness = deps.ledgerFreshness ?? sdkLedgerFreshness; return { name: 'burn__fingerprint', description: 'Cheap polling primitive over the burn ledger. Returns ' + '`{count}:{maxMtimeUnix}:{totalBytes}` joined by colons. ' + "Clients keep the last-seen value and skip re-querying when it's " + - 'unchanged. Optionally scoped to a session id or project path. Read-only.', + 'unchanged. The response also includes ledgerFreshness; check ' + + 'ledgerFreshness.stale before relying on ledger reads. Optionally scoped ' + + 'to a session id or project path. Read-only.', inputSchema: { type: 'object', properties: { @@ -61,8 +65,11 @@ export function createFingerprintTool(deps: FingerprintDeps = {}): ToolDefinitio const opts: { session?: string; project?: string } = {}; if (input.sessionId !== undefined) opts.session = input.sessionId; if (input.project !== undefined) opts.project = input.project; - const result = await callFingerprint(opts); - return result; + const [result, ledgerFreshness] = await Promise.all([ + callFingerprint(opts), + callLedgerFreshness(), + ]); + return { ...result, ledgerFreshness }; }, }; } diff --git a/packages/mcp/src/tools/session-cost.test.ts b/packages/mcp/src/tools/session-cost.test.ts index 5672031b..983e6e86 100644 --- a/packages/mcp/src/tools/session-cost.test.ts +++ b/packages/mcp/src/tools/session-cost.test.ts @@ -3,9 +3,14 @@ import { describe, it } from 'node:test'; import { createSessionCostTool, type SessionCostResult } from './session-cost.js'; +const FRESHNESS_DEP = { + ledgerFreshness: async () => ({ lastWriteAtMs: 1, staleAfterMs: 86_400_000, stale: false }), +}; + describe('createSessionCostTool', () => { it('returns the SDK no-session shape with the MCP-specific note when no id is registered', async () => { const tool = createSessionCostTool({ + ...FRESHNESS_DEP, defaultSessionId: undefined, sessionCost: async () => ({ sessionId: null, @@ -20,12 +25,34 @@ describe('createSessionCostTool', () => { assert.equal(result.sessionId, null); assert.equal(result.totalUSD, 0); assert.equal(result.turnCount, 0); + assert.deepEqual(result.ledgerFreshness, { + lastWriteAtMs: 1, + staleAfterMs: 86_400_000, + stale: false, + }); assert.match(result.note ?? '', /no session id provided and server was not registered/); }); + it('returns stale freshness metadata unchanged', async () => { + const tool = createSessionCostTool({ + ledgerFreshness: async () => ({ lastWriteAtMs: 1, staleAfterMs: 2, stale: true }), + defaultSessionId: 's1', + sessionCost: async () => ({ + sessionId: 's1', + totalUSD: 0, + totalTokens: 0, + turnCount: 0, + models: [], + }), + }); + const result = (await tool.handler({})) as SessionCostResult; + assert.deepEqual(result.ledgerFreshness, { lastWriteAtMs: 1, staleAfterMs: 2, stale: true }); + }); + it('uses the override sessionId when provided', async () => { let queriedFor: string | undefined; const tool = createSessionCostTool({ + ...FRESHNESS_DEP, defaultSessionId: 'default-id', sessionCost: async (opts) => { queriedFor = opts.session; @@ -45,6 +72,7 @@ describe('createSessionCostTool', () => { it('falls back to defaultSessionId when no override given', async () => { let queriedFor: string | undefined; const tool = createSessionCostTool({ + ...FRESHNESS_DEP, defaultSessionId: 'baked-id', sessionCost: async (opts) => { queriedFor = opts.session; @@ -63,6 +91,7 @@ describe('createSessionCostTool', () => { it('returns the SDK result verbatim when a session id is present', async () => { const tool = createSessionCostTool({ + ...FRESHNESS_DEP, defaultSessionId: 's1', sessionCost: async (opts) => ({ sessionId: opts.session ?? null, diff --git a/packages/mcp/src/tools/session-cost.ts b/packages/mcp/src/tools/session-cost.ts index 9ca57ade..f42472ec 100644 --- a/packages/mcp/src/tools/session-cost.ts +++ b/packages/mcp/src/tools/session-cost.ts @@ -1,5 +1,5 @@ -import { sessionCost as sdkSessionCost } from '@relayburn/sdk'; -import type { SessionCostResult as SdkSessionCostResult } from '@relayburn/sdk'; +import { ledgerFreshness as sdkLedgerFreshness, sessionCost as sdkSessionCost } from '@relayburn/sdk'; +import type { LedgerFreshness, SessionCostResult as SdkSessionCostResult } from '@relayburn/sdk'; import type { ToolDefinition } from '../types.js'; @@ -7,7 +7,7 @@ export interface SessionCostInput { sessionId?: string; } -export type SessionCostResult = SdkSessionCostResult; +export type SessionCostResult = SdkSessionCostResult & { ledgerFreshness: LedgerFreshness }; export interface SessionCostDeps { defaultSessionId: string | undefined; @@ -16,10 +16,12 @@ export interface SessionCostDeps { * exercise the tool surface without touching the on-disk ledger. */ sessionCost?: (opts: { session?: string }) => Promise; + ledgerFreshness?: () => Promise; } export function createSessionCostTool(deps: SessionCostDeps): ToolDefinition { const callSessionCost = deps.sessionCost ?? sdkSessionCost; + const callLedgerFreshness = deps.ledgerFreshness ?? sdkLedgerFreshness; return { name: 'burn__sessionCost', description: @@ -43,15 +45,18 @@ export function createSessionCostTool(deps: SessionCostDeps): ToolDefinition { const sessionId = input.sessionId ?? deps.defaultSessionId; const opts: { session?: string } = {}; if (sessionId !== undefined) opts.session = sessionId; - const result = await callSessionCost(opts); + const [result, ledgerFreshness] = await Promise.all([ + callSessionCost(opts), + callLedgerFreshness(), + ]); // The SDK's "no session id" note is generic ("no session id provided"); // keep the more descriptive variant the MCP tool used to surface so the // hint that the *server* should have been registered with one stays // visible to MCP clients. if (result.sessionId === null && sessionId === undefined) { - return { ...result, note: 'no session id provided and server was not registered with one' }; + return { ...result, ledgerFreshness, note: 'no session id provided and server was not registered with one' }; } - return result; + return { ...result, ledgerFreshness }; }, }; } diff --git a/packages/sdk-node/CHANGELOG.md b/packages/sdk-node/CHANGELOG.md index 666de257..4ccccbf9 100644 --- a/packages/sdk-node/CHANGELOG.md +++ b/packages/sdk-node/CHANGELOG.md @@ -2,6 +2,7 @@ ## [Unreleased] +- `ledgerFreshness()` exposes the ledger's last-write timestamp, configured threshold, and stale flag for Node and MCP presenters. - Cost calculations recognize Claude 5 and GPT-5.6 models, prefer first-party tariffs, and apply long-context price tiers. - `hotspots()` findings identify unknown pricing and rank unpriced sessions by token volume instead of $0.00. diff --git a/packages/sdk-node/src/binding.d.ts b/packages/sdk-node/src/binding.d.ts index 3fc04911..23ecf7e5 100644 --- a/packages/sdk-node/src/binding.d.ts +++ b/packages/sdk-node/src/binding.d.ts @@ -14,6 +14,7 @@ export declare class Ledger { export declare function ingest(opts?: unknown): Promise; export declare function summary(opts?: unknown): Promise; +export declare function ledgerFreshness(opts?: unknown): Promise; export declare function sessionCost(opts?: unknown): Promise; export declare function fingerprint(opts?: unknown): Promise; export declare function overhead(opts?: unknown): Promise; diff --git a/packages/sdk-node/src/index.cjs b/packages/sdk-node/src/index.cjs index df5f1419..8a4cac2e 100644 --- a/packages/sdk-node/src/index.cjs +++ b/packages/sdk-node/src/index.cjs @@ -87,6 +87,7 @@ module.exports = { Ledger, ingest: async (opts) => coerceBigInts(await binding.ingest(opts)), summary: async (opts) => coerceBigInts(await binding.summary(opts)), + ledgerFreshness: async (opts) => binding.ledgerFreshness(opts), sessionCost: async (opts) => coerceBigInts(await binding.sessionCost(opts)), fingerprint: async (opts) => coerceBigInts(await binding.fingerprint(opts)), overhead: async (opts) => coerceBigInts(await binding.overhead(opts)), diff --git a/packages/sdk-node/src/index.d.ts b/packages/sdk-node/src/index.d.ts index c532fb83..7c34c393 100644 --- a/packages/sdk-node/src/index.d.ts +++ b/packages/sdk-node/src/index.d.ts @@ -468,6 +468,18 @@ export declare function computeCompareExcluded( // stack (see issue #374), so there is nothing to log. // --------------------------------------------------------------------------- +export interface LedgerFreshnessOptions { ledgerHome?: string } +export interface LedgerFreshness { + /** Unix epoch milliseconds of the most recent ledger mutation. */ + lastWriteAtMs?: number; + /** Null when staleness warnings are disabled. */ + staleAfterMs: number | null; + stale: boolean; +} + +/** Inspect whether read/report data is older than the configured threshold. */ +export declare function ledgerFreshness(opts?: LedgerFreshnessOptions): Promise + export interface SearchQueryOptions { /** FTS5 query string. Phrase, boolean, and prefix syntax supported. */ query: string; diff --git a/packages/sdk-node/src/index.js b/packages/sdk-node/src/index.js index 95bc9771..63e695f8 100644 --- a/packages/sdk-node/src/index.js +++ b/packages/sdk-node/src/index.js @@ -100,6 +100,10 @@ export async function summary(opts) { return coerceBigInts(await binding.summary(opts)); } +export async function ledgerFreshness(opts) { + return binding.ledgerFreshness(opts); +} + export async function sessionCost(opts) { return coerceBigInts(await binding.sessionCost(opts)); } diff --git a/packages/sdk-node/test/conformance.test.js b/packages/sdk-node/test/conformance.test.js index 56b68722..8577401b 100644 --- a/packages/sdk-node/test/conformance.test.js +++ b/packages/sdk-node/test/conformance.test.js @@ -8,7 +8,15 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; -import { mkdtempSync, rmSync, cpSync, mkdirSync, readdirSync, readFileSync } from 'node:fs'; +import { + mkdtempSync, + rmSync, + cpSync, + mkdirSync, + readdirSync, + readFileSync, + writeFileSync, +} from 'node:fs'; import { tmpdir } from 'node:os'; import { join, resolve, dirname } from 'node:path'; import { fileURLToPath } from 'node:url'; @@ -62,6 +70,7 @@ test('sdk facade exposes the expected verb set', async (t) => { 'Ledger', 'ingest', 'summary', + 'ledgerFreshness', 'sessionCost', 'fingerprint', 'overhead', @@ -85,6 +94,12 @@ test('read verbs return stable shapes against the fixture ledger', async (t) => const ledgerHome = makeLedgerHome(); try { + const freshness = await sdk.ledgerFreshness({ ledgerHome }); + assert.equal(typeof freshness.stale, 'boolean'); + assert.ok(freshness.staleAfterMs === null || typeof freshness.staleAfterMs === 'number'); + assert.ok( + freshness.lastWriteAtMs === undefined || typeof freshness.lastWriteAtMs === 'number', + ); const summary = await sdk.summary({ ledgerHome }); assert.equal(typeof summary.totalCost, 'number'); assert.ok(Array.isArray(summary.byModel)); @@ -154,6 +169,24 @@ test('read verbs return stable shapes against the fixture ledger', async (t) => } }); +test('ledgerFreshness returns null threshold when warnings are disabled', async (t) => { + const sdk = await loadNapiSdk(t); + if (!sdk) return; + + const ledgerHome = makeLedgerHome(); + try { + writeFileSync( + join(ledgerHome, 'config.json'), + JSON.stringify({ staleness: { thresholdHours: -1 } }), + ); + const freshness = await sdk.ledgerFreshness({ ledgerHome }); + assert.equal(freshness.staleAfterMs, null); + assert.equal(freshness.stale, false); + } finally { + rmSync(ledgerHome, { recursive: true, force: true }); + } +}); + test('2.x extension verbs return stable shapes against the fixture ledger', async (t) => { const sdk = await loadNapiSdk(t); if (!sdk) return; diff --git a/tests/fixtures/cli-golden/snapshots/state-status-json.stdout.txt b/tests/fixtures/cli-golden/snapshots/state-status-json.stdout.txt index f36069be..eaeba160 100644 --- a/tests/fixtures/cli-golden/snapshots/state-status-json.stdout.txt +++ b/tests/fixtures/cli-golden/snapshots/state-status-json.stdout.txt @@ -21,7 +21,8 @@ "rows": 0 }, "archive": { - "schemaVersion": 5 + "schemaVersion": 7, + "lastWriteAtMs": "${TS}" }, "config": { "store": "off", diff --git a/tests/fixtures/cli-golden/snapshots/state-status.stdout.txt b/tests/fixtures/cli-golden/snapshots/state-status.stdout.txt index 36954c47..446da963 100644 --- a/tests/fixtures/cli-golden/snapshots/state-status.stdout.txt +++ b/tests/fixtures/cli-golden/snapshots/state-status.stdout.txt @@ -14,9 +14,10 @@ content DB (content.sqlite): path: ${RELAYBURN_HOME}/content.sqlite rows: 0 archive state: - schema version: 5 + schema version: 7 last built: never last rebuild: never + last write: ${TS} config: store: off retention: 90 days