Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions crates/tinyjuice-bus/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

pub mod names;
pub mod repl;
pub mod savings;
pub mod summary;
pub mod tools;
pub mod types;
Expand Down
4 changes: 4 additions & 0 deletions crates/tinyjuice-bus/src/lib_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,10 @@ fn the_contract_accepts_its_own_version_and_newer_minors() {
assert!(is_compatible(CONTRACT_VERSION));
assert!(is_compatible((CONTRACT_VERSION.0, CONTRACT_VERSION.1 + 1)));
assert!(!is_compatible((CONTRACT_VERSION.0 + 1, 0)));
assert!(
!is_compatible((2, 0)),
"savings members require contract 2.1"
);
}

#[test]
Expand Down
12 changes: 12 additions & 0 deletions crates/tinyjuice-bus/src/names.rs
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,14 @@ pub mod methods {
pub const QUERY: &str = "Query";
/// Convert HTML to Markdown inside the module.
pub const EXTRACT_HTML: &str = "ExtractHtml";
/// Sets the savings snapshot path and loads an existing snapshot once.
pub const CONFIGURE_SAVINGS: &str = "ConfigureSavings";
/// Records a saving with the host-resolved model input price.
pub const RECORD_SAVINGS: &str = "RecordSavings";
/// Returns the current persisted-format savings totals.
pub const SAVINGS_STATS: &str = "SavingsStats";
/// Clears savings totals and persists the empty snapshot.
pub const RESET_SAVINGS: &str = "ResetSavings";
/// Reports what the cache is holding.
pub const CACHE_STATS: &str = "CacheStats";
}
Expand All @@ -65,4 +73,8 @@ pub const METHODS: &[&str] = &[
methods::QUERY,
methods::EXTRACT_HTML,
methods::CACHE_STATS,
methods::CONFIGURE_SAVINGS,
methods::RECORD_SAVINGS,
methods::SAVINGS_STATS,
methods::RESET_SAVINGS,
];
48 changes: 48 additions & 0 deletions crates/tinyjuice-bus/src/savings.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
//! Persisted savings totals and requests shared with the compression module.

use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// Per-key (model / compressor) rolled-up savings.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SavingsBucket {
/// Number of shrinking compactions; saturates at `u64::MAX`.
pub events: u64,
/// Sum of estimated tokens before compaction.
pub original_tokens: u64,
/// Sum of estimated tokens after compaction.
pub compacted_tokens: u64,
/// Sum of positive pre/post token differences.
pub tokens_saved: u64,
/// Estimated dropped-input cost from host prices; saturates at `f64::MAX`.
pub cost_saved_usd: f64,
}

/// The full savings snapshot returned to callers / the dashboard.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SavingsAggregate {
/// Overall totals across every compaction.
pub total: SavingsBucket,
/// Breakdown by the model the savings were attributed to.
pub by_model: HashMap<String, SavingsBucket>,
/// Breakdown by which compressor produced the saving.
pub by_compressor: HashMap<String, SavingsBucket>,
}

/// One compaction attributed to the model that would consume its output.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RecordSavingsRequest {
/// Model identifier used for the dashboard breakdown and host price lookup.
pub model: String,
/// Compressor identifier used for the dashboard breakdown.
pub compressor: String,
/// Estimated input tokens before compaction.
pub original_tokens: u64,
/// Estimated input tokens after compaction.
pub compacted_tokens: u64,
/// Host-resolved price per million input tokens; absent means unpriced.
pub input_per_mtok_usd: Option<f64>,
}
2 changes: 1 addition & 1 deletion crates/tinyjuice-bus/src/version.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
//! The `TinyJuice` `TinyBus` contract version and its binding rule.

/// The wire contract version this crate defines.
pub const CONTRACT_VERSION: (u32, u32) = (2, 0);
pub const CONTRACT_VERSION: (u32, u32) = (2, 1);

/// Returns whether a host using [`CONTRACT_VERSION`] can bind to `module`.
///
Expand Down
1 change: 1 addition & 0 deletions crates/tinyjuice-module/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -21,4 +21,5 @@ tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
serde_json = "1"

[dev-dependencies]
tempfile = "3"
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] }
39 changes: 38 additions & 1 deletion crates/tinyjuice-module/src/service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -171,6 +171,32 @@ impl Compression {
let (entries, bytes) = tinyjuice::cache::stats();
Ok(CacheStats { entries, bytes })
}

/// Select and restore the module-owned savings snapshot once per path.
async fn configure_savings(&self, snapshot_path: String) -> BusResult<()> {
savings_ledger().configure(std::path::Path::new(&snapshot_path));
Ok(())
}

/// Accumulate a compaction using the host-provided model input price.
async fn record_savings(
&self,
request: tinyjuice_bus::savings::RecordSavingsRequest,
) -> BusResult<()> {
savings_ledger().record(&request);
Ok(())
}

/// Return a consistent copy of the module-owned savings totals.
async fn savings_stats(&self) -> BusResult<tinyjuice_bus::savings::SavingsAggregate> {
Ok(savings_ledger().stats())
}

/// Clear module-owned savings and persist the empty snapshot.
async fn reset_savings(&self) -> BusResult<()> {
savings_ledger().reset();
Ok(())
}
}

/// Requests may narrow a budget but never expand the module's stock ceilings.
Expand Down Expand Up @@ -275,7 +301,11 @@ mod exports {
"Repl",
"Query",
"ExtractHtml",
"CacheStats"
"CacheStats",
"ConfigureSavings",
"RecordSavings",
"SavingsStats",
"ResetSavings"
],
signals = [],
requires = [],
Expand All @@ -287,3 +317,10 @@ mod exports {
#[cfg(test)]
#[path = "service_tests.rs"]
mod tests;

/// One ledger per loaded module, shared by its compression service calls.
fn savings_ledger() -> &'static tinyjuice::savings::ledger::SavingsLedger {
static LEDGER: std::sync::OnceLock<tinyjuice::savings::ledger::SavingsLedger> =
std::sync::OnceLock::new();
LEDGER.get_or_init(tinyjuice::savings::ledger::SavingsLedger::new)
}
35 changes: 35 additions & 0 deletions crates/tinyjuice-module/tests/module_e2e.rs
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,7 @@ async fn the_built_module_compresses_and_recovers_over_a_real_broker() {
assert!(matches!(modules.list()[0].state, ModuleState::Ready));

compact_with_calls_back_to_the_host_for_a_focused_summary(&client, &proxy).await;
savings_are_computed_and_persisted_inside_the_loaded_module(&proxy).await;
broker_task.abort();
}

Expand Down Expand Up @@ -385,3 +386,37 @@ async fn typed_queries_execute_in_the_loaded_artifact(proxy: &tinybus::Proxy, to
assert!(markdown.contains("[details](/details)"));
assert!(!markdown.contains("secret()"));
}

async fn savings_are_computed_and_persisted_inside_the_loaded_module(proxy: &tinybus::Proxy) {
use tinyjuice_bus::names::methods;
use tinyjuice_bus::savings::{RecordSavingsRequest, SavingsAggregate};
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("savings.json");
proxy
.call::<()>(methods::CONFIGURE_SAVINGS, (path.to_str().unwrap(),))
.await
.unwrap();
proxy.call::<()>(methods::RESET_SAVINGS, ()).await.unwrap();
proxy
.call::<()>(
methods::RECORD_SAVINGS,
(RecordSavingsRequest {
model: "test-model".into(),
compressor: "generic".into(),
original_tokens: 2_000_000,
compacted_tokens: 1_000_000,
input_per_mtok_usd: Some(3.0),
},),
)
.await
.unwrap();
let stats: SavingsAggregate = proxy.call(methods::SAVINGS_STATS, ()).await.unwrap();
assert_eq!(stats.total.tokens_saved, 1_000_000);
assert_eq!(stats.by_model["test-model"].cost_saved_usd, 3.0);
let persisted: SavingsAggregate =
serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
assert_eq!(persisted.by_compressor["generic"].events, 1);
proxy.call::<()>(methods::RESET_SAVINGS, ()).await.unwrap();
let stats: SavingsAggregate = proxy.call(methods::SAVINGS_STATS, ()).await.unwrap();
assert_eq!(stats.total.events, 0);
}
28 changes: 28 additions & 0 deletions docs/specs/tinybus-module.md
Original file line number Diff line number Diff line change
Expand Up @@ -109,3 +109,31 @@ response characters, `max_line_chars=240` characters per returned line, and
narrow these values but cannot enlarge the module's stock ceilings. Grep keeps
matching lines before allocating remaining context slots. An explicit summary
`max_chars` also bounds a model-failure notice combined with its overview.

## Model-attributed savings (contract 2.1)

The module owns aggregation and best-effort JSON snapshot persistence. Hosts
link only `tinyjuice-bus::savings` and call these members:

- `ConfigureSavings(snapshot_path: String) -> ()` loads the snapshot once per
path. Repeating the same path preserves current in-memory totals.
- `RecordSavings(RecordSavingsRequest) -> ()` records a positive token saving.
The host supplies the attribution model, compressor identifier, pre/post
token estimates, and its catalog lookup of the input price per million
tokens. `None` records token savings with zero estimated dollar savings.
- `SavingsStats() -> SavingsAggregate` returns total, model, and compressor
breakdowns in the existing camelCase dashboard/snapshot format.
- `ResetSavings() -> ()` clears the totals and persists an empty snapshot.

The ledger does not call a hosted pricing service or depend on a host cost
catalog. A pricing callback remains host-owned; its result crosses the bus in
`RecordSavingsRequest::input_per_mtok_usd`. Invalid prices and overflowed
per-event costs are unpriced. Accumulated integer and dollar totals saturate
at `u64::MAX` and `f64::MAX`, respectively.
Counts are estimates, not provider-measured usage. The `SavingsRecord` event
callback surface remains available separately for richer accounting.

The snapshot path is trusted host configuration, like the cache disk root.
Missing or malformed snapshots retain the current totals, matching the former
OpenHuman ledger. Switching paths does not create an isolated per-user ledger;
hosts requiring that isolation must use separate module instances or reset.
3 changes: 3 additions & 0 deletions src/savings.rs
Original file line number Diff line number Diff line change
Expand Up @@ -212,3 +212,6 @@ pub fn record_event(record: SavingsRecord) {
#[cfg(test)]
#[path = "savings_tests.rs"]
mod tests;

/// Model-attributed totals and snapshot persistence behind the module contract.
pub mod ledger;
137 changes: 137 additions & 0 deletions src/savings/ledger.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,137 @@
//! Model-attributed savings aggregation and best-effort snapshot persistence.

use std::path::{Path, PathBuf};
use std::sync::Mutex;

pub use tinyjuice_bus::savings::{RecordSavingsRequest, SavingsAggregate, SavingsBucket};

#[derive(Default)]
struct State {
aggregate: SavingsAggregate,
snapshot_path: Option<PathBuf>,
}

/// A thread-safe ledger scoped to its caller, without a host pricing catalog.
///
/// Counts are estimates supplied by the compressor. Monetary savings use only
/// the host-provided input rate, with unpriced models reporting zero dollars.
#[derive(Default)]
pub struct SavingsLedger {
state: Mutex<State>,
}

impl SavingsLedger {
/// Creates an empty in-memory ledger.
#[must_use]
pub fn new() -> Self {
Self::default()
}

/// Selects a snapshot file and loads it once per path.
///
/// Repeated configuration of the same path does not reload stale disk
/// state. Missing or malformed snapshots leave the current totals intact,
/// matching the original host ledger. I/O failures are best-effort.
pub fn configure(&self, path: &Path) {
let mut state = self.state.lock().unwrap_or_else(|p| p.into_inner());
if state.snapshot_path.as_deref() == Some(path) {
return;
}
let loaded = std::fs::read_to_string(path)
.ok()
.and_then(|text| serde_json::from_str(&text).ok());
state.snapshot_path = Some(path.to_path_buf());
if let Some(aggregate) = loaded {
state.aggregate = aggregate;
}
}

/// Records a strictly smaller compaction and snapshots its totals.
///
/// Equal or expanded output is a no-op. Negative or non-finite input
/// prices and overflowed per-event costs are unpriced. Bucket totals
/// saturate at their largest finite representation rather than wrapping.
pub fn record(&self, request: &RecordSavingsRequest) {
if request.original_tokens <= request.compacted_tokens {
return;
}
let mut state = self.state.lock().unwrap_or_else(|p| p.into_inner());
let cost = request
.input_per_mtok_usd
.filter(|price| price.is_finite() && *price >= 0.0)
.map_or(0.0, |price| {
(request.original_tokens - request.compacted_tokens) as f64 / 1_000_000.0 * price
});
Comment thread
senamakel marked this conversation as resolved.
let cost = if cost.is_finite() { cost } else { 0.0 };
let aggregate = &mut state.aggregate;
add(&mut aggregate.total, request, cost);
add(
aggregate.by_model.entry(request.model.clone()).or_default(),
request,
cost,
);
add(
aggregate
.by_compressor
.entry(request.compressor.clone())
.or_default(),
request,
cost,
);
persist(&state);
}

/// Returns a consistent copy of all current totals.
#[must_use]
pub fn stats(&self) -> SavingsAggregate {
self.state
.lock()
.unwrap_or_else(|p| p.into_inner())
.aggregate
.clone()
}

/// Clears totals and writes the empty persisted snapshot.
pub fn reset(&self) {
let mut state = self.state.lock().unwrap_or_else(|p| p.into_inner());
state.aggregate = SavingsAggregate::default();
persist(&state);
}
}

/// Accumulate one event without wrapping persisted totals at their limits.
fn add(bucket: &mut SavingsBucket, request: &RecordSavingsRequest, cost: f64) {
bucket.events = bucket.events.saturating_add(1);
bucket.original_tokens = bucket
.original_tokens
.saturating_add(request.original_tokens);
bucket.compacted_tokens = bucket
.compacted_tokens
.saturating_add(request.compacted_tokens);
bucket.tokens_saved = bucket
.tokens_saved
.saturating_add(request.original_tokens - request.compacted_tokens);
bucket.cost_saved_usd = (bucket.cost_saved_usd + cost).min(f64::MAX);
}
Comment thread
senamakel marked this conversation as resolved.

/// Write the snapshot best-effort, retaining the former host I/O behavior.
fn persist(state: &State) {
let Some(path) = state.snapshot_path.as_ref() else {
return;
};
if let Some(parent) = path.parent() {
let _ = std::fs::create_dir_all(parent);
}
match serde_json::to_string(&state.aggregate) {
Ok(json) => {
if let Err(error) = std::fs::write(path, json) {
log::debug!("[tokenjuice][savings] snapshot write failed: {error}");
}
}
Err(error) => log::debug!("[tokenjuice][savings] snapshot serialize failed: {error}"),
}
}

#[cfg(test)]
#[path = "ledger_tests.rs"]
mod tests;
Loading
Loading