You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/09/23 09:34:06 UTC
(maka) 03/09: feat(pricing): add revisioned native rate catalog
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch feat/runtime-host-rust
in repository https://gitbox.apache.org/repos/asf/maka.git
commit defd2ed756c047f90bf9ab13c02a9735aa918771
Author: Wang <m4...@gmail.com>
AuthorDate: Wed Sep 23 15:38:22 2026 +0800
feat(pricing): add revisioned native rate catalog
Generated-by: Codex
---
crates/config/build.rs | 1 +
crates/config/migrations/0001_configuration.sql | 11 +
crates/config/src/database.rs | 1 +
crates/config/src/lib.rs | 1 +
crates/config/src/pricing.rs | 168 +++++++++++++
crates/config/src/pricing/catalog.rs | 92 +++++++
crates/config/tests/pricing.rs | 253 ++++++++++++++++++++
crates/protocol/src/lib.rs | 1 +
crates/protocol/src/pricing.rs | 136 +++++++++++
crates/runtime-host/src/server.rs | 1 +
crates/runtime-host/src/server/authority.rs | 2 +
crates/runtime-host/src/server/dispatch.rs | 3 +
crates/runtime-host/src/server/operations.rs | 14 ++
crates/runtime-host/src/server/pricing.rs | 55 +++++
.../tests/integration/client_pricing.rs} | 49 ++--
crates/runtime-host/tests/integration/main.rs | 1 +
crates/runtime/src/lib.rs | 1 +
crates/runtime/src/pricing.rs | 266 +++++++++++++++++++++
docs/rust-parity.md | 2 +-
docs/rust-parity.zh-CN.md | 2 +-
docs/rust-runtime.md | 4 +-
docs/rust-runtime.zh-CN.md | 3 +-
scripts/rust/catalog-facts-entry.mjs | 5 +
scripts/rust/generate-catalog-facts.mjs | 19 ++
tests/fixtures/client-pricing-entry.mjs | 120 ++++++++++
tests/fixtures/client.mjs | 1 +
26 files changed, 1178 insertions(+), 34 deletions(-)
diff --git a/crates/config/build.rs b/crates/config/build.rs
index 682d044606..47c4973eac 100644
--- a/crates/config/build.rs
+++ b/crates/config/build.rs
@@ -23,6 +23,7 @@ fn main() {
"../../scripts/rust/generate-catalog-facts.mjs",
"../../scripts/rust/catalog-facts-entry.mjs",
"../../packages/core/src",
+ "../../packages/runtime/src/telemetry/builtin-pricing.ts",
"../../scripts/sync-model-metadata.mjs",
"../../scripts/model-metadata/models-dev-api.snapshot.json",
"../../package-lock.json",
diff --git a/crates/config/migrations/0001_configuration.sql b/crates/config/migrations/0001_configuration.sql
index a0646847bc..1c7deca824 100644
--- a/crates/config/migrations/0001_configuration.sql
+++ b/crates/config/migrations/0001_configuration.sql
@@ -23,6 +23,17 @@ CREATE TABLE connection_catalog (
default_target TEXT
);
INSERT INTO connection_catalog VALUES(1, 0, NULL);
+CREATE TABLE pricing_authority (
+ singleton INTEGER PRIMARY KEY CHECK(singleton = 1),
+ revision INTEGER NOT NULL CHECK(revision BETWEEN 0 AND 9007199254740991),
+ builtin_digest TEXT
+);
+INSERT INTO pricing_authority VALUES(1, 0, NULL);
+CREATE TABLE pricing_overrides (
+ model_key TEXT PRIMARY KEY,
+ sort_key BLOB NOT NULL,
+ record_json TEXT NOT NULL CHECK(json_valid(record_json) AND length(CAST(record_json AS BLOB)) <= 4096)
+);
CREATE TABLE connections (
connection_id TEXT PRIMARY KEY,
slug TEXT NOT NULL UNIQUE,
diff --git a/crates/config/src/database.rs b/crates/config/src/database.rs
index c97d0b406c..8c3fb2584e 100644
--- a/crates/config/src/database.rs
+++ b/crates/config/src/database.rs
@@ -139,5 +139,6 @@ async fn initialize(connection: &mut SqliteConnection) -> std::result::Result<()
)
.execute(&mut *connection)
.await?;
+ crate::pricing::initialize(connection).await?;
Ok(())
}
diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs
index ff6bb02863..6ffc73c06c 100644
--- a/crates/config/src/lib.rs
+++ b/crates/config/src/lib.rs
@@ -33,6 +33,7 @@ pub mod onboarding;
pub mod plugin_authorization;
mod plugin_credentials;
mod policy;
+pub mod pricing;
pub mod projection;
mod vault;
diff --git a/crates/config/src/pricing.rs b/crates/config/src/pricing.rs
new file mode 100644
index 0000000000..39bd98efa7
--- /dev/null
+++ b/crates/config/src/pricing.rs
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+mod catalog;
+pub(crate) use catalog::initialize;
+
+use crate::{ConfigError, ConfigurationStore, Result, TransactionMode, database::unsigned};
+use maka_runtime::pricing::{Entry, Mutation, Page, Pricing, Query, ResetEffect, Update, Updated};
+use sqlx::{Row, SqliteConnection};
+
+const MAX_REVISION: u64 = 9_007_199_254_740_991;
+const PAGE_ITEMS: usize = 128;
+// Reserve envelope space; each admitted record is bounded independently.
+const PAGE_ENTRY_BYTES: usize = 47 * 1024;
+
+impl ConfigurationStore {
+ /// Continuations are valid only while both bundled and custom rates are unchanged.
+ pub async fn query_pricing(&self, query: Query) -> Result<Page> {
+ self.transaction(TransactionMode::Deferred, move |connection| Box::pin(async move {
+ let revision = revision(connection).await?;
+ let offset = match query {
+ Query::Start => 0,
+ Query::Continue { revision: expected, offset } => {
+ if expected > MAX_REVISION || offset > i64::MAX as u64 {
+ return Err(invalid("invalid pricing continuation"));
+ }
+ if expected != revision {
+ return Ok(Page::RevisionChanged { expected_revision: expected, actual_revision: revision });
+ }
+ let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM effective_pricing")
+ .fetch_one(&mut *connection).await?;
+ if offset >= unsigned(count)? {
+ return Err(invalid("pricing offset is outside the catalog"));
+ }
+ offset
+ }
+ };
+ let rows = sqlx::query(
+ "SELECT record_json, custom, builtin FROM effective_pricing ORDER BY sort_key LIMIT ? OFFSET ?"
+ ).bind((PAGE_ITEMS + 1) as i64).bind(offset as i64)
+ .fetch_all(&mut *connection).await?;
+ let mut entries = Vec::new();
+ let mut bytes = 0;
+ let mut more = false;
+ for row in rows {
+ let pricing = serde_json::from_str(row.try_get("record_json")?)?;
+ let entry = if row.try_get::<bool, _>("custom")? {
+ Entry::Custom { pricing, reset_effect: if row.try_get::<bool, _>("builtin")? {
+ ResetEffect::RestoreBuiltin
+ } else { ResetEffect::BecomeUnpriced } }
+ } else { Entry::Builtin { pricing } };
+ let size = serde_json::to_vec(&entry)?.len() + 1;
+ if entries.len() == PAGE_ITEMS || bytes + size > PAGE_ENTRY_BYTES {
+ more = true;
+ break;
+ }
+ bytes += size;
+ entries.push(entry);
+ }
+ let next_offset = more.then_some(offset + entries.len() as u64);
+ Ok(Page::Page { revision, offset, entries, next_offset })
+ })).await
+ }
+
+ /// A quote is a value, not a live lookup. Persist it with the model accounting fact.
+ pub async fn model_pricing(&self, model_key: String) -> Result<(u64, Option<Pricing>)> {
+ maka_runtime::pricing::validate_key(&model_key).map_err(invalid)?;
+ self.transaction(TransactionMode::Deferred, move |connection| {
+ Box::pin(async move {
+ let revision = revision(connection).await?;
+ let record: Option<String> = sqlx::query_scalar(
+ "SELECT record_json FROM effective_pricing WHERE model_key = ?",
+ )
+ .bind(model_key)
+ .fetch_optional(&mut *connection)
+ .await?;
+ Ok((
+ revision,
+ record
+ .map(|record| serde_json::from_str(&record))
+ .transpose()?,
+ ))
+ })
+ })
+ .await
+ }
+
+ pub async fn update_pricing(&self, update: Update) -> Result<Updated> {
+ if update.expected_revision > MAX_REVISION {
+ return Err(invalid("invalid pricing revision"));
+ }
+ match &update.mutation {
+ Mutation::Upsert { pricing } => pricing.validate().map_err(invalid)?,
+ Mutation::Delete { model_key } => {
+ maka_runtime::pricing::validate_key(model_key).map_err(invalid)?
+ }
+ }
+ self.transaction(TransactionMode::Immediate, move |connection| Box::pin(async move {
+ let current = revision(connection).await?;
+ if current != update.expected_revision {
+ return Ok(Updated::RevisionConflict {
+ expected_revision: update.expected_revision, actual_revision: current,
+ });
+ }
+ let changed = match update.mutation {
+ Mutation::Upsert { pricing } => {
+ let old: Option<String> = sqlx::query_scalar(
+ "SELECT record_json FROM pricing_overrides WHERE model_key = ?"
+ ).bind(&pricing.model_key).fetch_optional(&mut *connection).await?;
+ if old.as_deref().map(serde_json::from_str::<Pricing>).transpose()?.as_ref() == Some(&pricing) {
+ false
+ } else {
+ let record = serde_json::to_string(&pricing)?;
+ if record.len() > 4096 { return Err(invalid("model price exceeds record limit")); }
+ sqlx::query(
+ "INSERT INTO pricing_overrides(model_key, sort_key, record_json) VALUES(?, ?, ?)
+ ON CONFLICT(model_key) DO UPDATE SET record_json = excluded.record_json"
+ ).bind(&pricing.model_key).bind(sort_key(&pricing.model_key)).bind(record)
+ .execute(&mut *connection).await?;
+ true
+ }
+ }
+ Mutation::Delete { model_key } => sqlx::query(
+ "DELETE FROM pricing_overrides WHERE model_key = ?"
+ ).bind(model_key).execute(&mut *connection).await?.rows_affected() != 0,
+ };
+ if !changed { return Ok(Updated::Unchanged { revision: current }); }
+ let revision = current.checked_add(1).filter(|revision| *revision <= MAX_REVISION)
+ .ok_or_else(|| invalid("pricing revision exhausted"))?;
+ sqlx::query("UPDATE pricing_authority SET revision = ? WHERE singleton = 1")
+ .bind(revision as i64).execute(&mut *connection).await?;
+ Ok(Updated::Committed { revision })
+ })).await
+ }
+}
+
+async fn revision(connection: &mut SqliteConnection) -> Result<u64> {
+ unsigned(
+ sqlx::query_scalar("SELECT revision FROM pricing_authority WHERE singleton = 1")
+ .fetch_one(connection)
+ .await?,
+ )
+}
+
+fn invalid(message: impl Into<String>) -> ConfigError {
+ ConfigError::Invalid(message.into())
+}
+
+// Big-endian UTF-16 preserves the client's exact, locale-independent key ordering.
+fn sort_key(key: &str) -> Vec<u8> {
+ key.encode_utf16().flat_map(u16::to_be_bytes).collect()
+}
diff --git a/crates/config/src/pricing/catalog.rs b/crates/config/src/pricing/catalog.rs
new file mode 100644
index 0000000000..c111becf2c
--- /dev/null
+++ b/crates/config/src/pricing/catalog.rs
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+//! Disposable bundled-rate projection. Only overrides and the revision are durable.
+use super::{MAX_REVISION, sort_key};
+use maka_event_log::StoreError;
+use maka_runtime::pricing::Pricing;
+use sqlx::{Connection, QueryBuilder, SqliteConnection};
+use std::sync::LazyLock;
+
+const FACTS: &str = include_str!(concat!(env!("OUT_DIR"), "/pricing-facts.json"));
+const DIGEST: &str = include_str!(concat!(env!("OUT_DIR"), "/pricing-digest.txt"));
+static PRICES: LazyLock<Vec<Pricing>> = LazyLock::new(|| {
+ let prices: Vec<Pricing> = serde_json::from_str(FACTS).expect("generated model rates");
+ for price in &prices {
+ price.validate().expect("valid generated model rate");
+ }
+ prices
+});
+
+pub(crate) async fn initialize(connection: &mut SqliteConnection) -> Result<(), StoreError> {
+ let mut tx = connection.begin_with("BEGIN IMMEDIATE").await?;
+ sqlx::raw_sql(
+ "CREATE TEMP TABLE builtin_pricing (
+ model_key TEXT PRIMARY KEY, sort_key BLOB NOT NULL, record_json TEXT NOT NULL
+ );
+ CREATE TEMP VIEW effective_pricing AS
+ SELECT b.model_key, b.sort_key, b.record_json, 0 AS custom, 1 AS builtin
+ FROM builtin_pricing b WHERE NOT EXISTS (
+ SELECT 1 FROM main.pricing_overrides p WHERE p.model_key = b.model_key
+ )
+ UNION ALL
+ SELECT p.model_key, p.sort_key, p.record_json, 1 AS custom,
+ EXISTS(SELECT 1 FROM builtin_pricing b WHERE b.model_key = p.model_key) AS builtin
+ FROM main.pricing_overrides p;",
+ )
+ .execute(&mut *tx)
+ .await?;
+ for chunk in PRICES.chunks(128) {
+ let mut query =
+ QueryBuilder::new("INSERT INTO builtin_pricing(model_key, sort_key, record_json) ");
+ query.push_values(chunk, |mut row, pricing| {
+ row.push_bind(&pricing.model_key)
+ .push_bind(sort_key(&pricing.model_key))
+ .push_bind(serde_json::to_string(pricing).expect("finite generated rates"));
+ });
+ query.build().execute(&mut *tx).await?;
+ }
+ // An application update can change bundled rates without changing overrides.
+ // Invalidate previously issued cursors and distinguish newly captured quotes.
+ let changed = sqlx::query(
+ "UPDATE pricing_authority
+ SET revision = revision + CASE WHEN builtin_digest IS NULL THEN 0 ELSE 1 END,
+ builtin_digest = ?
+ WHERE singleton = 1 AND builtin_digest IS NOT ?
+ AND (builtin_digest IS NULL OR revision < ?)",
+ )
+ .bind(DIGEST)
+ .bind(DIGEST)
+ .bind(MAX_REVISION as i64)
+ .execute(&mut *tx)
+ .await?;
+ if changed.rows_affected() == 0 {
+ let digest: String =
+ sqlx::query_scalar("SELECT builtin_digest FROM pricing_authority WHERE singleton = 1")
+ .fetch_one(&mut *tx)
+ .await?;
+ if digest != DIGEST {
+ return Err(StoreError::InvalidTransition(
+ "pricing revision exhausted".into(),
+ ));
+ }
+ }
+ tx.commit().await?;
+ Ok(())
+}
diff --git a/crates/config/tests/pricing.rs b/crates/config/tests/pricing.rs
new file mode 100644
index 0000000000..33332d5721
--- /dev/null
+++ b/crates/config/tests/pricing.rs
@@ -0,0 +1,253 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use maka_config::ConfigurationStore;
+use maka_event_log::root::{RootNamespaces, RootOwner};
+use maka_runtime::pricing::{Entry, Mutation, Page, Pricing, Query, ResetEffect, Update, Updated};
+use std::sync::Arc;
+
+async fn entries(store: &ConfigurationStore) -> (u64, Vec<Entry>) {
+ let mut query = Query::Start;
+ let mut all = Vec::new();
+ loop {
+ let page = store.query_pricing(query).await.unwrap();
+ assert!(serde_json::to_vec(&page).unwrap().len() <= 48 * 1024);
+ let Page::Page {
+ revision,
+ offset,
+ entries,
+ next_offset,
+ } = page
+ else {
+ panic!("stable snapshot")
+ };
+ assert_eq!(offset, all.len() as u64);
+ assert!(entries.len() <= 128);
+ all.extend(entries);
+ match next_offset {
+ Some(offset) => query = Query::Continue { revision, offset },
+ None => return (revision, all),
+ }
+ }
+}
+
+fn price(entry: &Entry) -> &Pricing {
+ match entry {
+ Entry::Builtin { pricing } | Entry::Custom { pricing, .. } => pricing,
+ }
+}
+
+#[tokio::test]
+async fn pricing_cas_pages_and_quotes_survive_reopen_and_bundled_rate_replacement() {
+ let temp = tempfile::tempdir().unwrap();
+ let owner = Arc::new(
+ RootOwner::create(
+ &temp.path().join("root"),
+ &RootNamespaces {
+ ownership: temp.path().join("owners"),
+ control: temp.path().join("control"),
+ },
+ )
+ .unwrap(),
+ );
+ let store = ConfigurationStore::for_root(owner.clone()).await.unwrap();
+ let (revision, initial) = entries(&store).await;
+ assert_eq!(revision, 0);
+ assert!(
+ initial.len() > 128,
+ "exercise pagination against the real offline catalog"
+ );
+ let original = price(&initial[0]).clone();
+ let mut custom = original.clone();
+ custom.input_usd_per_million += 10.0;
+ let update = Update {
+ expected_revision: 0,
+ mutation: Mutation::Upsert {
+ pricing: custom.clone(),
+ },
+ };
+ assert_eq!(
+ store.update_pricing(update.clone()).await.unwrap(),
+ Updated::Committed { revision: 1 }
+ );
+ // A lost mutation reply cannot overwrite a later state on retry.
+ assert_eq!(
+ store.update_pricing(update).await.unwrap(),
+ Updated::RevisionConflict {
+ expected_revision: 0,
+ actual_revision: 1
+ }
+ );
+ assert!(matches!(
+ store
+ .query_pricing(Query::Continue {
+ revision: 0,
+ offset: 128
+ })
+ .await
+ .unwrap(),
+ Page::RevisionChanged {
+ expected_revision: 0,
+ actual_revision: 1
+ }
+ ));
+ assert_eq!(
+ store
+ .update_pricing(Update {
+ expected_revision: 1,
+ mutation: Mutation::Upsert {
+ pricing: custom.clone()
+ },
+ })
+ .await
+ .unwrap(),
+ Updated::Unchanged { revision: 1 }
+ );
+ let captured = store.model_pricing(custom.model_key.clone()).await.unwrap();
+ assert_eq!(captured, (1, Some(custom.clone())));
+ assert!(entries(&store).await.1.iter().any(|entry| *entry
+ == Entry::Custom {
+ pricing: custom.clone(),
+ reset_effect: ResetEffect::RestoreBuiltin,
+ }));
+
+ // Supplementary-plane keys sort before U+E000 in JS, unlike UTF-8 ordering.
+ let mut revision = 1;
+ for key in ["fixture:\u{e000}", "fixture:\u{10000}"] {
+ revision += 1;
+ assert_eq!(
+ store
+ .update_pricing(Update {
+ expected_revision: revision - 1,
+ mutation: Mutation::Upsert {
+ pricing: Pricing {
+ model_key: key.into(),
+ ..custom.clone()
+ }
+ },
+ })
+ .await
+ .unwrap(),
+ Updated::Committed { revision }
+ );
+ }
+ let (_, all) = entries(&store).await;
+ assert!(all.windows(2).all(|pair| {
+ price(&pair[0])
+ .model_key
+ .encode_utf16()
+ .cmp(price(&pair[1]).model_key.encode_utf16())
+ .is_lt()
+ }));
+ assert!(
+ all.iter()
+ .filter(|entry| matches!(
+ entry,
+ Entry::Custom {
+ reset_effect: ResetEffect::BecomeUnpriced,
+ ..
+ }
+ ))
+ .count()
+ >= 2
+ );
+ assert!(
+ store
+ .query_pricing(Query::Continue {
+ revision,
+ offset: all.len() as u64
+ })
+ .await
+ .is_err()
+ );
+ store.close().await.unwrap();
+
+ let store = ConfigurationStore::for_root(owner.clone()).await.unwrap();
+ assert_eq!(
+ store.model_pricing(custom.model_key.clone()).await.unwrap(),
+ (3, Some(custom.clone()))
+ );
+ assert_eq!(
+ store
+ .update_pricing(Update {
+ expected_revision: 3,
+ mutation: Mutation::Delete {
+ model_key: custom.model_key.clone()
+ }
+ })
+ .await
+ .unwrap(),
+ Updated::Committed { revision: 4 }
+ );
+ assert_eq!(
+ store.model_pricing(custom.model_key.clone()).await.unwrap(),
+ (4, Some(original))
+ );
+ assert_eq!(
+ captured,
+ (1, Some(custom)),
+ "captured quote is not a mutable pricing reference"
+ );
+ assert_eq!(
+ store
+ .update_pricing(Update {
+ expected_revision: 4,
+ mutation: Mutation::Delete {
+ model_key: "missing:model".into()
+ }
+ })
+ .await
+ .unwrap(),
+ Updated::Unchanged { revision: 4 }
+ );
+ store.close().await.unwrap();
+
+ // Emulate a binary built from another bundled-rate version; keep overrides.
+ let db = rusqlite::Connection::open(owner.canonical_path().join("configuration-rust.sqlite"))
+ .unwrap();
+ db.execute(
+ "UPDATE pricing_authority SET builtin_digest = 'previous-build'",
+ [],
+ )
+ .unwrap();
+ drop(db);
+ let store = ConfigurationStore::for_root(owner).await.unwrap();
+ assert!(matches!(
+ store
+ .query_pricing(Query::Continue {
+ revision: 4,
+ offset: 1
+ })
+ .await
+ .unwrap(),
+ Page::RevisionChanged {
+ actual_revision: 5,
+ ..
+ }
+ ));
+ assert_eq!(
+ store
+ .model_pricing("fixture:\u{e000}".into())
+ .await
+ .unwrap()
+ .0,
+ 5
+ );
+ store.close().await.unwrap();
+}
diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs
index 45e1fdfcb5..68656a668e 100644
--- a/crates/protocol/src/lib.rs
+++ b/crates/protocol/src/lib.rs
@@ -40,6 +40,7 @@ pub mod onboarding;
pub mod operation;
pub mod operation_error;
pub mod plugin;
+pub mod pricing;
pub mod project;
pub mod request_headers;
pub mod resource;
diff --git a/crates/protocol/src/pricing.rs b/crates/protocol/src/pricing.rs
new file mode 100644
index 0000000000..f6dd6a9329
--- /dev/null
+++ b/crates/protocol/src/pricing.rs
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Operation, ProtocolError, Result, codec::MAX_SAFE_INTEGER};
+use maka_runtime::pricing::{Mutation, Page, Query, Update, Updated};
+use serde_json::Value;
+
+pub fn supports(operation: Operation) -> bool {
+ matches!(
+ operation,
+ Operation::PricingQuery | Operation::PricingMutate
+ )
+}
+
+pub enum Input {
+ Query(Query),
+ Update(Update),
+}
+
+pub fn decode_input(operation: Operation, value: &Value) -> Result<Input> {
+ match operation {
+ Operation::PricingQuery => {
+ let query: Query = decode(value)?;
+ if let Query::Continue { revision, offset } = query
+ && (revision > MAX_SAFE_INTEGER || offset > MAX_SAFE_INTEGER)
+ {
+ return Err(ProtocolError::invalid("invalid pricing continuation"));
+ }
+ Ok(Input::Query(query))
+ }
+ Operation::PricingMutate => {
+ let input: Update = decode(value)?;
+ if input.expected_revision > MAX_SAFE_INTEGER {
+ return Err(ProtocolError::invalid("invalid pricing revision"));
+ }
+ match &input.mutation {
+ Mutation::Upsert { pricing } => pricing.validate(),
+ Mutation::Delete { model_key } => maka_runtime::pricing::validate_key(model_key),
+ }
+ .map_err(ProtocolError::invalid)?;
+ Ok(Input::Update(input))
+ }
+ _ => Err(ProtocolError::invalid("unknown pricing operation")),
+ }
+}
+
+pub fn decode_output(operation: Operation, value: &Value) -> Result<Value> {
+ match operation {
+ Operation::PricingQuery => {
+ let result: Page = decode(value)?;
+ match result {
+ Page::RevisionChanged {
+ expected_revision,
+ actual_revision,
+ } => {
+ conflict(expected_revision, actual_revision)?;
+ }
+ Page::Page {
+ revision,
+ offset,
+ entries,
+ next_offset,
+ } => {
+ if revision > MAX_SAFE_INTEGER
+ || offset > MAX_SAFE_INTEGER
+ || entries.len() > 128
+ || serde_json::to_vec(value).map_err(invalid)?.len() > 48 * 1024
+ || next_offset.is_some_and(|next| {
+ entries.is_empty()
+ || next > MAX_SAFE_INTEGER
+ || next != offset + entries.len() as u64
+ })
+ {
+ return Err(ProtocolError::invalid("invalid pricing page"));
+ }
+ let mut previous: Option<&str> = None;
+ for entry in &entries {
+ let (maka_runtime::pricing::Entry::Builtin { pricing }
+ | maka_runtime::pricing::Entry::Custom { pricing, .. }) = entry;
+ pricing.validate().map_err(ProtocolError::invalid)?;
+ if previous.is_some_and(|key| {
+ key.encode_utf16()
+ .cmp(pricing.model_key.encode_utf16())
+ .is_ge()
+ }) {
+ return Err(ProtocolError::invalid("pricing page is not ordered"));
+ }
+ previous = Some(&pricing.model_key);
+ }
+ }
+ }
+ }
+ Operation::PricingMutate => match decode::<Updated>(value)? {
+ Updated::Committed { revision } | Updated::Unchanged { revision } => {
+ if revision > MAX_SAFE_INTEGER {
+ return Err(ProtocolError::invalid("invalid pricing revision"));
+ }
+ }
+ Updated::RevisionConflict {
+ expected_revision,
+ actual_revision,
+ } => conflict(expected_revision, actual_revision)?,
+ },
+ _ => return Err(ProtocolError::invalid("unknown pricing operation")),
+ }
+ Ok(value.clone())
+}
+
+fn conflict(expected: u64, actual: u64) -> Result<()> {
+ if expected > MAX_SAFE_INTEGER || actual > MAX_SAFE_INTEGER || expected == actual {
+ return Err(ProtocolError::invalid("invalid pricing revision conflict"));
+ }
+ Ok(())
+}
+fn decode<T: serde::de::DeserializeOwned>(value: &Value) -> Result<T> {
+ serde_json::from_value(value.clone()).map_err(invalid)
+}
+fn invalid(error: impl std::fmt::Display) -> ProtocolError {
+ ProtocolError::invalid(error.to_string())
+}
diff --git a/crates/runtime-host/src/server.rs b/crates/runtime-host/src/server.rs
index aaa362b90f..c2c93bdebe 100644
--- a/crates/runtime-host/src/server.rs
+++ b/crates/runtime-host/src/server.rs
@@ -43,6 +43,7 @@ mod operations;
mod outbound;
pub(crate) mod plugin_authorization;
mod plugin_remote;
+mod pricing;
mod projects;
mod sandbox_setup;
pub(crate) use projects::Usage as ProjectUsage;
diff --git a/crates/runtime-host/src/server/authority.rs b/crates/runtime-host/src/server/authority.rs
index d4223f5baa..cf02d24615 100644
--- a/crates/runtime-host/src/server/authority.rs
+++ b/crates/runtime-host/src/server/authority.rs
@@ -223,6 +223,8 @@ fn path_free(request: &Request) -> bool {
| Operation::ConnectionCatalogQuery
| Operation::ExecutorCatalogQuery
| Operation::ModelProviderCatalogQuery
+ | Operation::PricingQuery
+ | Operation::PricingMutate
| Operation::ConnectionModelsFetch
| Operation::ConnectionOnboardingVerify
| Operation::ConnectionOnboardingSave
diff --git a/crates/runtime-host/src/server/dispatch.rs b/crates/runtime-host/src/server/dispatch.rs
index 8de10d8903..4dde118976 100644
--- a/crates/runtime-host/src/server/dispatch.rs
+++ b/crates/runtime-host/src/server/dispatch.rs
@@ -82,6 +82,9 @@ impl Host {
self.plugins.wake_background_work();
return Ok(Outcome::success(serde_json::json!({})));
}
+ if maka_protocol::pricing::supports(operation) {
+ return super::pricing::execute(self, operation, &input).await;
+ }
if operation == Operation::HostUpgradePrepare {
return match self
.prepare_retirement(
diff --git a/crates/runtime-host/src/server/operations.rs b/crates/runtime-host/src/server/operations.rs
index df741b3ce3..e2b1fb94c4 100644
--- a/crates/runtime-host/src/server/operations.rs
+++ b/crates/runtime-host/src/server/operations.rs
@@ -30,6 +30,10 @@ pub struct Operations;
impl OperationRegistry for Operations {
fn decode_input(&self, operation: Operation, value: &Value) -> Result<Value> {
+ if maka_protocol::pricing::supports(operation) {
+ maka_protocol::pricing::decode_input(operation, value)?;
+ return Ok(value.clone());
+ }
if maka_protocol::sandbox_setup::supports(operation) {
maka_protocol::sandbox_setup::decode_input(value)?;
return Ok(value.clone());
@@ -111,6 +115,9 @@ impl OperationRegistry for Operations {
}
}
fn decode_output(&self, operation: Operation, value: &Value) -> Result<Value> {
+ if maka_protocol::pricing::supports(operation) {
+ return maka_protocol::pricing::decode_output(operation, value);
+ }
if maka_protocol::sandbox_setup::supports(operation) {
maka_protocol::sandbox_setup::decode_output(value)?;
return Ok(value.clone());
@@ -187,6 +194,13 @@ impl OperationRegistry for Operations {
}
}
fn error_codes(&self, operation: Operation) -> Option<&[OperationErrorCode]> {
+ if maka_protocol::pricing::supports(operation) {
+ return Some(if operation == Operation::PricingQuery {
+ configuration::QUERY_ERRORS
+ } else {
+ configuration::MUTATION_ERRORS
+ });
+ }
if maka_protocol::sandbox_setup::supports(operation) {
return Some(maka_protocol::sandbox_setup::ERRORS);
}
diff --git a/crates/runtime-host/src/server/pricing.rs b/crates/runtime-host/src/server/pricing.rs
new file mode 100644
index 0000000000..94dbcf5b2a
--- /dev/null
+++ b/crates/runtime-host/src/server/pricing.rs
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use super::{Host, HostError, configuration};
+use maka_protocol::{Operation, Outcome, pricing::Input};
+use maka_runtime::pricing::Updated;
+use serde_json::{Value, json};
+use std::sync::atomic::Ordering;
+
+pub(super) async fn execute(
+ host: &Host,
+ operation: Operation,
+ value: &Value,
+) -> Result<Outcome, HostError> {
+ let output = match maka_protocol::pricing::decode_input(operation, value)? {
+ Input::Query(input) => host
+ .configuration
+ .query_pricing(input)
+ .await
+ .map(serde_json::to_value),
+ Input::Update(input) => host
+ .configuration
+ .update_pricing(input)
+ .await
+ .map(|result| {
+ if matches!(result, Updated::Committed { .. }) {
+ let revision = host.change_revision.fetch_add(1, Ordering::SeqCst) + 1;
+ let _ = host
+ .changes
+ .send(json!({ "kind": "configuration.changed", "revision": revision }));
+ }
+ serde_json::to_value(result)
+ }),
+ };
+ Ok(match output {
+ Ok(value) => Outcome::success(maka_protocol::pricing::decode_output(operation, &value?)?),
+ Err(error) => Outcome::failure(configuration::failure(error)),
+ })
+}
diff --git a/crates/runtime/src/lib.rs b/crates/runtime-host/tests/integration/client_pricing.rs
similarity index 57%
copy from crates/runtime/src/lib.rs
copy to crates/runtime-host/tests/integration/client_pricing.rs
index 886f4d8f54..ad645696d2 100644
--- a/crates/runtime/src/lib.rs
+++ b/crates/runtime-host/tests/integration/client_pricing.rs
@@ -17,33 +17,22 @@
* under the License.
*/
-//! Runtime facts and execution contracts; no V8 or SQLite dependency.
-pub mod access;
-pub mod archive;
-pub mod artifact;
-pub mod attachment;
-pub mod capability;
-pub mod composition;
-pub mod configuration;
-pub mod context;
-pub mod continuation;
-pub mod event;
-mod event_write;
-pub mod execution;
-pub mod executor;
-pub mod handoff;
-pub mod input;
-pub mod interaction;
-pub mod message;
-pub mod model;
-pub mod oauth;
-pub mod provider;
-pub mod read;
-pub mod scope;
-pub mod session;
-pub mod shell_result;
-pub mod shell_run;
-pub mod terminal;
-pub mod tool_call;
-pub mod tool_output;
-pub mod tools;
+use super::support::client_probe::ClientFixture;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn source_client_pages_edits_and_recovers_native_pricing() {
+ let fixture = ClientFixture::new("maka-pricing-client-");
+ for reopened in [false, true] {
+ fixture
+ .run(
+ "--pricing-workspace",
+ reopened,
+ if reopened {
+ "pricing-reopened"
+ } else {
+ "pricing-passed"
+ },
+ )
+ .await;
+ }
+}
diff --git a/crates/runtime-host/tests/integration/main.rs b/crates/runtime-host/tests/integration/main.rs
index 345ce0d1f5..6891cd9ac8 100644
--- a/crates/runtime-host/tests/integration/main.rs
+++ b/crates/runtime-host/tests/integration/main.rs
@@ -35,6 +35,7 @@ mod client_onboarding;
mod client_openai_options;
mod client_patch;
mod client_plugins;
+mod client_pricing;
mod client_questions;
mod client_remote_access;
mod client_shell;
diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs
index 886f4d8f54..22fea9930a 100644
--- a/crates/runtime/src/lib.rs
+++ b/crates/runtime/src/lib.rs
@@ -37,6 +37,7 @@ pub mod interaction;
pub mod message;
pub mod model;
pub mod oauth;
+pub mod pricing;
pub mod provider;
pub mod read;
pub mod scope;
diff --git a/crates/runtime/src/pricing.rs b/crates/runtime/src/pricing.rs
new file mode 100644
index 0000000000..5227124f11
--- /dev/null
+++ b/crates/runtime/src/pricing.rs
@@ -0,0 +1,266 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+//! Non-secret model rates and auditable estimates. This is not a billing ledger.
+
+use crate::model::ModelUsage;
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
+#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
+pub enum Query {
+ Start,
+ Continue { revision: u64, offset: u64 },
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[serde(
+ tag = "source",
+ rename_all = "snake_case",
+ rename_all_fields = "camelCase",
+ deny_unknown_fields
+)]
+pub enum Entry {
+ Builtin {
+ pricing: Pricing,
+ },
+ Custom {
+ pricing: Pricing,
+ reset_effect: ResetEffect,
+ },
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum ResetEffect {
+ RestoreBuiltin,
+ BecomeUnpriced,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
+#[serde(
+ tag = "kind",
+ rename_all = "snake_case",
+ rename_all_fields = "camelCase",
+ deny_unknown_fields
+)]
+pub enum Page {
+ Page {
+ revision: u64,
+ offset: u64,
+ entries: Vec<Entry>,
+ next_offset: Option<u64>,
+ },
+ RevisionChanged {
+ expected_revision: u64,
+ actual_revision: u64,
+ },
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
+#[serde(
+ tag = "kind",
+ rename_all = "snake_case",
+ rename_all_fields = "camelCase",
+ deny_unknown_fields
+)]
+pub enum Mutation {
+ Upsert { pricing: Pricing },
+ Delete { model_key: String },
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct Update {
+ pub expected_revision: u64,
+ pub mutation: Mutation,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[serde(
+ tag = "kind",
+ rename_all = "snake_case",
+ rename_all_fields = "camelCase",
+ deny_unknown_fields
+)]
+pub enum Updated {
+ Committed {
+ revision: u64,
+ },
+ Unchanged {
+ revision: u64,
+ },
+ RevisionConflict {
+ expected_revision: u64,
+ actual_revision: u64,
+ },
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct Pricing {
+ pub model_key: String,
+ #[serde(rename = "inputUsdPer1M")]
+ pub input_usd_per_million: f64,
+ #[serde(rename = "outputUsdPer1M")]
+ pub output_usd_per_million: f64,
+ #[serde(
+ rename = "cacheReadUsdPer1M",
+ default,
+ deserialize_with = "present_rate",
+ skip_serializing_if = "Option::is_none"
+ )]
+ #[schemars(with = "f64")]
+ pub cache_read_usd_per_million: Option<f64>,
+ #[serde(
+ rename = "cacheWriteUsdPer1M",
+ default,
+ deserialize_with = "present_rate",
+ skip_serializing_if = "Option::is_none"
+ )]
+ #[schemars(with = "f64")]
+ pub cache_write_usd_per_million: Option<f64>,
+}
+
+impl Pricing {
+ pub fn validate(&self) -> Result<(), &'static str> {
+ validate_key(&self.model_key)?;
+ for rate in [
+ Some(self.input_usd_per_million),
+ Some(self.output_usd_per_million),
+ self.cache_read_usd_per_million,
+ self.cache_write_usd_per_million,
+ ]
+ .into_iter()
+ .flatten()
+ {
+ if !rate.is_finite() || rate < 0.0 {
+ return Err("invalid model price");
+ }
+ }
+ Ok(())
+ }
+
+ /// Input and output totals include their cache and reasoning subsets.
+ /// Optional cache rates inherit the input rate, never a made-up discount.
+ /// Missing total usage or an unrepresentable estimate is not zero cost.
+ pub fn estimate(&self, usage: &ModelUsage) -> Result<Option<f64>, &'static str> {
+ self.validate()?;
+ let (Some(input), Some(output)) = (usage.input_tokens, usage.output_tokens) else {
+ return Ok(None);
+ };
+ if input != 0
+ && ((usage.cache_read_tokens.is_none()
+ && self
+ .cache_read_usd_per_million
+ .is_some_and(|rate| rate != self.input_usd_per_million))
+ || (usage.cache_write_tokens.is_none()
+ && self
+ .cache_write_usd_per_million
+ .is_some_and(|rate| rate != self.input_usd_per_million)))
+ {
+ return Ok(None);
+ }
+ let read = usage.cache_read_tokens.unwrap_or(0).min(input);
+ let write = usage.cache_write_tokens.unwrap_or(0).min(input - read);
+ let uncached = input - read - write;
+ // Scale the count before multiplication: a finite rate need not fit when
+ // multiplied by a whole token count but may still produce a finite cost.
+ let amount = uncached as f64 / 1_000_000.0 * self.input_usd_per_million
+ + read as f64 / 1_000_000.0
+ * self
+ .cache_read_usd_per_million
+ .unwrap_or(self.input_usd_per_million)
+ + write as f64 / 1_000_000.0
+ * self
+ .cache_write_usd_per_million
+ .unwrap_or(self.input_usd_per_million)
+ + output as f64 / 1_000_000.0 * self.output_usd_per_million;
+ Ok(amount.is_finite().then_some(amount))
+ }
+}
+
+/// Model keys are exact identities, not normalized provider or model names.
+pub fn validate_key(key: &str) -> Result<(), &'static str> {
+ if key.is_empty()
+ || key.trim() != key
+ || key.encode_utf16().count() > 128
+ || key.chars().any(char::is_control)
+ {
+ return Err("invalid pricing model key");
+ }
+ Ok(())
+}
+
+fn present_rate<'de, D: serde::Deserializer<'de>>(
+ deserializer: D,
+) -> Result<Option<f64>, D::Error> {
+ f64::deserialize(deserializer).map(Some)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn estimate_preserves_unknowns_and_does_not_charge_token_subsets_twice() {
+ let price = Pricing {
+ model_key: "provider:model".into(),
+ input_usd_per_million: 10.0,
+ output_usd_per_million: 20.0,
+ cache_read_usd_per_million: Some(1.0),
+ cache_write_usd_per_million: Some(12.0),
+ };
+ let mut usage = ModelUsage {
+ input_tokens: Some(1_000_000),
+ output_tokens: Some(1_000_000),
+ cache_read_tokens: Some(200_000),
+ cache_write_tokens: Some(100_000),
+ reasoning_tokens: Some(500_000),
+ };
+ assert_eq!(price.estimate(&usage).unwrap(), Some(28.4));
+ usage.input_tokens = None;
+ assert_eq!(price.estimate(&usage).unwrap(), None);
+ usage.input_tokens = Some(0);
+ usage.output_tokens = Some(0);
+ assert_eq!(price.estimate(&usage).unwrap(), Some(0.0));
+ // An impossible cache report cannot produce a negative uncached charge.
+ usage.input_tokens = Some(100_000);
+ assert_eq!(price.estimate(&usage).unwrap(), Some(0.1));
+ usage.cache_read_tokens = None;
+ assert_eq!(
+ price.estimate(&usage).unwrap(),
+ None,
+ "unknown cache split changes the price"
+ );
+ let huge = Pricing {
+ input_usd_per_million: f64::MAX,
+ cache_read_usd_per_million: None,
+ cache_write_usd_per_million: None,
+ ..price
+ };
+ usage.cache_read_tokens = None;
+ usage.cache_write_tokens = None;
+ usage.input_tokens = Some(1);
+ assert!(huge.estimate(&usage).unwrap().unwrap().is_finite());
+ usage.input_tokens = Some(u64::MAX);
+ assert_eq!(huge.estimate(&usage).unwrap(), None);
+ }
+}
diff --git a/docs/rust-parity.md b/docs/rust-parity.md
index e3602edb9e..29d6e26381 100644
--- a/docs/rust-parity.md
+++ b/docs/rust-parity.md
@@ -89,7 +89,7 @@ implementations; wire adapters may retain existing client vocabulary.
| Goal | Query/arm/control, continuation, termination, budget and recovery semantics. | **Plugin + Host.** Goal policy owns subsequent submissions; Host enforces admitted hard limits and records usage. Retiring the plugin must close future submission admission. |
| Daily review / recap | Daily-review query/mutate, scheduled review and `session.recap.generate`. | **Plugin + Host.** Summarization/selection/output in plugins; reuse Scheduler and authorized history/model services. Host retains any canonical Session metadata commit. |
| External agents | Setup start/query/cancel; execution adapters, configuration, auth, conversation identity, adapter-specific attachments/interactions/resume/fork. | **Plugin + Host.** Implement concrete CLI/ACP adapters as Executor plugins over owned processes/HTTP. Generic Executor support is not a shipped adapter. Host owns authorization, cancellation and canonical external-event recording. |
-| Usage / Pricing | Physical Agent and auxiliary SDK call accounting and bounded source reads are implemented. Public Usage queries, revision-consistent screens/activity pages, pricing query/mutate and valuation remain. | **Plugin + Host.** Reports, price policy and rebuildable projections may be an Insights domain; Host records usage independently of plugin availability and provides consistent reads. Missing usage must not become zero. |
+| Usage / Pricing | Physical Agent and auxiliary SDK accounting, bounded source reads, and native pricing query/CAS edits are implemented. Public plugin accounting/pricing capabilities, revision-consistent Usage screens/activity pages and per-call valuation remain. | **Plugin + Host.** Reports and rebuildable projections may be an Insights domain; Host records usage independently of plugin availability and provides consistent reads. Missing usage must not become zero. |
| Background health | BackgroundTaskHealth process and endpoint checks. | **Plugin + Host.** Health interpretation and Tool in a plugin; Host exposes authorized resource observations and bounded probes. A stored PID is not resource ownership. |
| Session lineage | Public Rust/JS history copying and source reads, native branch/revision create/abandon, independent inherited pruning, Desktop canonical-input editing and complete durable drafts are implemented. Regeneration is removed; edits use revision and resend. | **Host.** One transactional lineage and workspace authority; plugins request changes through commands, not private lineage storage. |
| Session lifecycle | Native and public plugin removal/preview/receipt APIs, atomic family retirement and queued cancellation, restart cleanup and shared-worktree ownership are implemented. Unreferenced canonical-material reclamation and shared Session queries remain. | **Host.** Preserve referenced history and accepted receipts; unknown process cleanup must not delete its workspace. Shared queries require real collaboration authorization. |
diff --git a/docs/rust-parity.zh-CN.md b/docs/rust-parity.zh-CN.md
index c4fe08de40..db16ffdfc2 100644
--- a/docs/rust-parity.zh-CN.md
+++ b/docs/rust-parity.zh-CN.md
@@ -79,7 +79,7 @@ runtime 契约不能反向依赖插件实现，协议适配层可以保留现有
| Goal | 查询、arm、控制、续跑、终止、预算与恢复语义。 | **插件 + Host。** Goal 决定后续提交；Host 执行已准入的硬限制并记录用量。插件退休后不能继续提交。 |
| Daily review／recap | daily-review 查询／修改、定时复盘、`session.recap.generate`。 | **插件 + Host。** 选择、总结及输出由插件负责，复用 Scheduler 和授权历史／模型服务；规范 Session 元数据的提交仍归 Host。 |
| 外部 agent | setup start/query/cancel；具体执行适配、配置、鉴权、对话身份，以及附件／交互／resume／fork。 | **插件 + Host。** CLI／ACP 适配作为 Executor 插件，使用受管理进程／HTTP。已有 Executor 框架不等于已有具体 adapter。Host 负责授权、取消和外部事件落盘。 |
-| Usage／Pricing | 已实现 Agent 与辅助 SDK 调用的物理准入计量及有界事实读取。公共 Usage 查询、一致版本视图及活动分页、Pricing 查询／修改和估价仍待接入。 | **插件 + Host。** 报表、价格策略和可重建投影可归 Insights 领域；Host 不依赖插件存活来记录用量，并提供一致快照。缺失用量不能视为零。 |
+| Usage／Pricing | 已实现 Agent 与辅助 SDK 的物理计量、有界事实读取及原生报价查询／CAS 修改。公共插件计量／报价能力、一致版本的 Usage 视图与活动分页、逐调用估价仍待接入。 | **插件 + Host。** 报表和可重建投影可归 Insights 领域；Host 不依赖插件存活来记录用量，并提供一致快照。缺失用量不能视为零。 |
| 后台健康 | BackgroundTaskHealth 的进程和端点检查。 | **插件 + Host。** 插件解释健康状态并提供工具；Host 提供授权资源观察和有界探测。保存 PID 不等于拥有进程。 |
| Session 谱系 | 已实现公共 Rust／JS 历史复制和原始输入读取、原生分支／修订创建与放弃、继承历史独立裁剪、Desktop 规范输入编辑及完整持久草稿。已移除 regenerate，修改请求使用编辑重发。 | **Host。** 谱系及工作区只有一个事务权威；插件请求命令，不在私有存储中重做谱系。 |
| Session 生命周期 | 已实现原生与公共插件删除／预览／回执接口、原子家族退休及队列取消、重启清理和共享 worktree 所有权。无引用规范材料回收及 shared 查询仍待完成。 | **Host。** 保留被引用历史及已接受回执；进程清理不确定时不得删除工作目录。shared 查询需要真实协作授权。 |
diff --git a/docs/rust-runtime.md b/docs/rust-runtime.md
index 033433529e..98311a05ed 100644
--- a/docs/rust-runtime.md
+++ b/docs/rust-runtime.md
@@ -215,7 +215,9 @@ user/project content paths stay separate from private journals.
- Accounting counts physical model admissions, including failed retries and auxiliary
plugin calls. Provider usage survives rejected output; missing counters and unresolved
outcomes remain unknown. History copies do not duplicate usage, and Session removal
- retains accounting facts. Usage screens and pricing are not yet connected.
+ retains accounting facts. Pricing queries and CAS edits use a bounded, revision-pinned
+ catalog; custom rates survive restart and bundled-rate updates invalidate old cursors.
+ Usage screens and per-call valuation are not yet connected.
- Subscriptions deliver frames only after `subscription.ready`. Reconnection replays
active text from the committed log under backpressure; no copied transcript overlay.
- Model messages, content and tool outcomes are typed through provider projection.
diff --git a/docs/rust-runtime.zh-CN.md b/docs/rust-runtime.zh-CN.md
index 024d12fd68..5c49af6090 100644
--- a/docs/rust-runtime.zh-CN.md
+++ b/docs/rust-runtime.zh-CN.md
@@ -168,7 +168,8 @@ Host 不可连接不代表进程已停止。`host logs --root-id <rootId>` 返
失败响应的片段只用于展示，不纳入模型历史；用户取消不显示为 provider 失败。
- 计量按模型物理准入统计，包含失败重试和插件辅助调用。被拒绝的回复仍保留已报告用量；
缺失计数与未确认结果保持未知。历史复制不重复计量，Session 删除保留计量事实。
- Usage 页面与 Pricing 尚未接入。
+ 报价支持有界、固定版本的查询与 CAS 修改；自定义报价跨重启保留，内置报价更新使旧分页失效。
+ Usage 页面与逐调用估价尚未接入。
- 模型消息、内容块与工具结果在 provider 投影中保持类型化；路由与发现共享类型化契约。
工具 JSON、schema 和厂商扩展保留开放结构。
- 订阅收到 `subscription.ready` 后才交付帧。重连按背压从已提交日志补发活动文本，
diff --git a/scripts/rust/catalog-facts-entry.mjs b/scripts/rust/catalog-facts-entry.mjs
index 6fd7551073..2f979840aa 100644
--- a/scripts/rust/catalog-facts-entry.mjs
+++ b/scripts/rust/catalog-facts-entry.mjs
@@ -32,6 +32,11 @@ import {
buildConnectionModelCatalogEntries,
} from '../../packages/core/src/model-catalog.ts';
import { getModelCapabilities } from '@ai-sdk/anthropic/internal';
+import { BUILTIN_PRICING } from '../../packages/runtime/src/telemetry/builtin-pricing.ts';
+
+export function outputPricingFacts() {
+ return BUILTIN_PRICING;
+}
export function outputProviderFacts() {
return Object.fromEntries(
diff --git a/scripts/rust/generate-catalog-facts.mjs b/scripts/rust/generate-catalog-facts.mjs
index d1c48b161b..d1cdcd5bbe 100644
--- a/scripts/rust/generate-catalog-facts.mjs
+++ b/scripts/rust/generate-catalog-facts.mjs
@@ -18,6 +18,7 @@
*/
import { createRequire } from 'node:module';
+import { createHash } from 'node:crypto';
import { readFile, writeFile } from 'node:fs/promises';
import { isAbsolute, join, resolve } from 'node:path';
import { fileURLToPath, pathToFileURL } from 'node:url';
@@ -26,6 +27,7 @@ const directory = resolve(process.argv[2]);
const require = createRequire(join(resolve(process.env.MAKA_JS_DEPS || root), 'package.json'));
const { build } = require('esbuild');
const metadata = join(directory, 'model-metadata.generated.ts');
+const pricing = join(directory, 'model-pricing.generated.ts');
const { main } = await import('../sync-model-metadata.mjs');
await main([
'node',
@@ -34,6 +36,8 @@ await main([
join(root, 'scripts/model-metadata/models-dev-api.snapshot.json'),
'--output',
metadata,
+ '--pricing-output',
+ pricing,
]);
const bundle = join(directory, 'catalog-facts-source.mjs');
const result = await build({
@@ -52,6 +56,9 @@ const result = await build({
build.onResolve({ filter: /^\.\/model-metadata\.generated\.js$/ }, () => ({
path: metadata,
}));
+ build.onResolve({ filter: /^\.\/model-pricing\.generated\.js$/ }, () => ({
+ path: pricing,
+ }));
build.onResolve({ filter: /^@maka\// }, async ({ path }) => {
const [, name, ...subpath] = path.split('/');
const packageRoot = join(root, 'packages', name);
@@ -89,3 +96,15 @@ await writeFile(
JSON.stringify(source.outputProviderFacts()),
);
await writeFile(join(directory, 'catalog-oracle.json'), JSON.stringify(source.oracleFixtures()));
+const pricingFacts = JSON.stringify(
+ source
+ .outputPricingFacts()
+ .sort((left, right) =>
+ left.modelKey < right.modelKey ? -1 : left.modelKey > right.modelKey ? 1 : 0,
+ ),
+);
+await writeFile(join(directory, 'pricing-facts.json'), pricingFacts);
+await writeFile(
+ join(directory, 'pricing-digest.txt'),
+ createHash('sha256').update(pricingFacts).digest('hex'),
+);
diff --git a/tests/fixtures/client-pricing-entry.mjs b/tests/fixtures/client-pricing-entry.mjs
new file mode 100644
index 0000000000..a8af2b64bd
--- /dev/null
+++ b/tests/fixtures/client-pricing-entry.mjs
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import assert from 'node:assert/strict';
+import { connect } from 'node:net';
+import { once } from 'node:events';
+import { parseArgs } from 'node:util';
+import { connectRuntimeHostMessageTransport } from '../../packages/runtime-host/src/client/connection.ts';
+import { FramedTransport } from '../../packages/runtime-host/src/transport/framed-transport.ts';
+import {
+ INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID,
+ RUNTIME_HOST_PROTOCOL_VERSION,
+} from '../../packages/runtime-host/src/protocol/index.ts';
+
+const { values } = parseArgs({
+ options: {
+ socket: { type: 'string' },
+ 'root-id': { type: 'string' },
+ 'pricing-workspace': { type: 'string' },
+ reopened: { type: 'boolean' },
+ },
+});
+const socket = connect(values.socket);
+const transport = new FramedTransport(socket);
+let connection;
+try {
+ await once(socket, 'connect');
+ const connected = await connectRuntimeHostMessageTransport({
+ transport,
+ expectedRootId: values['root-id'],
+ compositionId: INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID,
+ protocol: { min: RUNTIME_HOST_PROTOCOL_VERSION, max: RUNTIME_HOST_PROTOCOL_VERSION },
+ handshakeTimeoutMs: 3000,
+ livenessIntervalMs: 60000,
+ });
+ assert.equal(connected.kind, 'connected');
+ connection = connected.connection;
+ const request = (operation, input) => connection.request(operation, input, 5000);
+ const snapshot = async () => {
+ let page = await request('pricing.query', { kind: 'start' });
+ const revision = page.revision;
+ const entries = [];
+ while (true) {
+ assert.equal(page.kind, 'page');
+ assert.equal(page.revision, revision);
+ assert.equal(page.offset, entries.length);
+ entries.push(...page.entries);
+ if (page.nextOffset === null) return { revision, entries };
+ page = await request('pricing.query', {
+ kind: 'continue',
+ revision,
+ offset: page.nextOffset,
+ });
+ }
+ };
+ const initial = await snapshot();
+ assert(initial.entries.length > 128);
+ const key = 'fixture:priced-model';
+ const pricing = { modelKey: key, inputUsdPer1M: 2.5, outputUsdPer1M: 10 };
+ if (values.reopened) {
+ assert.equal(initial.revision, 1);
+ assert.deepEqual(
+ initial.entries.find(({ pricing }) => pricing.modelKey === key),
+ {
+ source: 'custom',
+ pricing,
+ resetEffect: 'become_unpriced',
+ },
+ );
+ assert.deepEqual(
+ await request('pricing.mutate', {
+ expectedRevision: 1,
+ mutation: { kind: 'delete', modelKey: key },
+ }),
+ { kind: 'committed', revision: 2 },
+ );
+ assert(!(await snapshot()).entries.some(({ pricing }) => pricing.modelKey === key));
+ } else {
+ assert.equal(initial.revision, 0);
+ const mutation = { expectedRevision: 0, mutation: { kind: 'upsert', pricing } };
+ assert.deepEqual(await request('pricing.mutate', mutation), { kind: 'committed', revision: 1 });
+ assert.deepEqual(await request('pricing.mutate', mutation), {
+ kind: 'revision_conflict',
+ expectedRevision: 0,
+ actualRevision: 1,
+ });
+ assert.deepEqual(
+ await request('pricing.query', {
+ kind: 'continue',
+ revision: 0,
+ offset: 128,
+ }),
+ { kind: 'revision_changed', expectedRevision: 0, actualRevision: 1 },
+ );
+ assert.deepEqual(await request('pricing.mutate', { ...mutation, expectedRevision: 1 }), {
+ kind: 'unchanged',
+ revision: 1,
+ });
+ }
+ console.log(values.reopened ? 'pricing-reopened' : 'pricing-passed');
+} finally {
+ transport.abort();
+ await connection?.close();
+}
diff --git a/tests/fixtures/client.mjs b/tests/fixtures/client.mjs
index be69ab7a22..45ae42a4aa 100644
--- a/tests/fixtures/client.mjs
+++ b/tests/fixtures/client.mjs
@@ -22,6 +22,7 @@ import { spawnSync } from 'node:child_process';
import { withSourceBundle } from '../support/source.mjs';
const entries = [
+ ['--pricing-workspace', 'client-pricing-entry.mjs'],
['--sandbox-workspace', 'client-sandbox-entry.mjs'],
['--skills-client-workspace', 'client-skills-entry.mjs'],
['--scheduler-workspace', 'client-scheduler-entry.mjs'],
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org