diff --git a/mesh/SPEC.md b/mesh/SPEC.md index 2321f97d5..ab3a187ca 100644 --- a/mesh/SPEC.md +++ b/mesh/SPEC.md @@ -570,6 +570,76 @@ These hold a serve to its record. Whether such a record carries any weight in a given dispute is not this specification's to say. It is evidence anyone can produce and check without the steward's help, opening apart. +### §6.5 The weights change record + +A receipt's `weights_digest` (§3.6) names the weights that answered. Without a +further record, nothing shows when a backend's weights changed, from what, or +whether the change was one its steward had planned for. A backend's **change +record** is that record. It is a hash-chained sequence of entries, each sealed +as a JCR-1 envelope under the steward key that seals its receipts, with the +profile tag `mesh_change: "mesh/1"`. + +Every entry carries `backend`, `recipe`, `seq` (its position, from 0), `prev` +(the content id of the envelope before it, absent for the first), `made_at` +(Unix seconds) and `kind`. Optional fields are omitted when absent. There are +two kinds: + +| `kind` | Fields | Records | +|---|---|---| +| `plan` | `precisions`, `fidelity_floor`, `validation`, `foresees`, `risk_assessment`? | the changes the steward foresees for the backend, before them | +| `weights` | `from`?, `to`, `precision`?, `fidelity`, `plan`?, `validation_result`?, `effective_from`, `reason` | one change of the backend's weights | + +The digests are hex content addresses. For a `weights` entry: +- `from` and `to` are the weights before and after the change; `from` is absent + for the backend's first entry of this kind. +- `plan` is the content id of the `plan` entry the change falls under, and is + absent for a change no plan foresaw. +- `effective_from` is the first sequence of the steward's receipt log (§6.3) + served on the new weights. + +An entry MUST be refused when: + +- it is for another backend, or does not chain: its `seq` is not the next, or + its `prev` is not the content id of the envelope before it; +- its `made_at` is earlier than the entry before it; +- it is a `weights` entry and: + - its `from` is not the `to` of the last `weights` entry before it; + - its `to` equals its `from`; + - its `effective_from` is earlier than the last change's; +- it names a `plan`, and that is not a `plan` entry earlier in the same record, + or the change is outside it. A change is outside its plan when it has: + - another `recipe`; + - a `precision` the plan's non-empty `precisions` does not list, or no + precision at all; + - a `fidelity` below the plan's `fidelity_floor`; + - no `validation_result`. + +A plan therefore covers only the changes recorded after it. A change recorded +as planned was foreseen before it was made, by the record's own order. A plan +recorded later does not reach back over it. A change no plan covers is +recorded too, without a `plan`. The record shows what changed, and makes no +judgement of it. + +**The gate.** Once a backend has a change record, the steward MUST serve on it +only the weights its last `weights` entry has in effect. It MUST refuse a serve +before any grant is decided or work runs where the engine names other weights, +or none. It MUST refuse to seal a serve whose engine reports other weights, or +none. A weights change takes effect only by being recorded, so every receipt +from `effective_from` on was served on the weights the change put in effect. +A steward that restarts holds a backend to the record it persisted, once every +entry verifies under its key, the record holds, and its weights in effect are +the ones the engine runs. + +**Coverage.** A party holding a receipt and the record for its backend reads the +`weights` entry in effect at the receipt's `sequence`: the last one whose +`effective_from` is at or before it. The receipt is covered when that entry is +for its backend and recipe, its `to` is the receipt's `weights_digest`, and, +where both name one, its precision is the receipt's. Otherwise it is not +covered. The entry then says which change put the weights in effect, when the +steward made it, under which plan, and with which validation result. A position +in the receipt log orders a change against receipts without trusting a clock; +`made_at` is the steward's statement of the time. + --- ## §7 Conformance @@ -594,6 +664,8 @@ A conforming implementation MUST: 5. Seal the receipt as a JCR-1 `mesh/1` envelope and compute `uj_per_accepted_token` (§6). 6. Maintain a tamper-evident hash chain (§6.3). +7. Where a backend has a change record, refuse its entries and its serves as + §6.5 sets out, so that its weights change only by being recorded. The published vectors in [`../conformance/v1/`](../conformance/v1/) pin a deterministic reference serve. `reference/signed_core.hex` MUST canonical-CBOR diff --git a/mesh/mesh-rs/crates/mesh-core/src/lib.rs b/mesh/mesh-rs/crates/mesh-core/src/lib.rs index 3623a78b9..eeec59713 100644 --- a/mesh/mesh-rs/crates/mesh-core/src/lib.rs +++ b/mesh/mesh-rs/crates/mesh-core/src/lib.rs @@ -43,9 +43,10 @@ pub use journal::{FileJournal, Journal, JournalEntry, JournalRecord, MemoryJourn pub use presentation::Presentation; use mesh_engine::ServingEngine; use mesh_meter::{Meter, Session}; +use mesh_receipt::change::{ChangeBody, ChangeEntry, ChangeLog, ChangePlan, WeightsChange}; use mesh_receipt::{output_commitment, payload_digest, MeshReceipt, ReceiptError, ReceiptLog}; use mesh_router::{BackendSpec, Router}; -use mesh_schema::{Energy, RouteDecision, ServeOutcome, ServeRequest}; +use mesh_schema::{BackendId, Energy, RouteDecision, ServeOutcome, ServeRequest}; /// The JCP capability action a Mesh serve request exercises. pub const INFER_ACTION: &str = "infer"; @@ -112,6 +113,13 @@ pub enum MeshError { /// The chosen backend has no registered engine (internal invariant). #[error("internal: no engine registered for backend `{0}`")] NoEngine(String), + /// The backend has a change record and runs weights the record does not + /// have in effect (SPEC §6.5): its weights were changed some other way. + #[error("backend `{backend}` runs weights {running}, and its change record has {recorded} in effect")] + WeightsUnrecorded { backend: String, running: String, recorded: String }, + /// A change record that does not hold, or a change it refuses (SPEC §6.5). + #[error("change record: {0}")] + ChangeRecord(String), /// A backend reported delivering above the level it is rated for. The /// router admitted it on the rating, so it cannot claim a higher basis /// after the fact. @@ -180,6 +188,11 @@ pub struct Mesh<'m> { /// steward that restarts can disclose the opening of any receipt it /// sealed, without having stored one. output_key: [u8; 32], + /// Each backend's weights change record, for the backends that have one + /// (SPEC §6.5). + changes: HashMap, + /// The steward key, which change records are sealed under as receipts are. + change_key: SigningKey, } impl<'m> Mesh<'m> { @@ -190,6 +203,7 @@ impl<'m> Mesh<'m> { pub fn new(meter: &'m dyn Meter, steward_sk: SigningKey, steward: impl Into) -> Self { let steward_did = Did::from_verifying_key(&steward_sk.verifying_key()); let output_key = ContentId::of_bytes(&[b"mesh/1 output opening key\0".as_slice(), &steward_sk.to_bytes()].concat()).0; + let change_key = steward_sk.clone(); Self { router: Router::new(), engines: HashMap::new(), @@ -205,6 +219,8 @@ impl<'m> Mesh<'m> { unresolved: Vec::new(), clock: system_clock, output_key, + changes: HashMap::new(), + change_key, } } @@ -362,12 +378,114 @@ impl<'m> Mesh<'m> { } /// Register a backend: its routing spec and the engine that serves it. + /// + /// Registering another engine for a backend that has a change record does + /// not change the weights it may serve: until the change is recorded + /// ([`record_weights`](Self::record_weights)), it is refused. pub fn register(&mut self, spec: BackendSpec, engine: Box) -> &mut Self { self.engines.insert(spec.id.0.clone(), engine); self.router.register(spec); self } + /// Change a registered backend's weights to `engine`'s, recording the + /// change in the backend's change record first (SPEC §6.5), and return + /// the change's id. The first call for a backend records the weights it + /// starts on and gives it a change record. From then on the backend serves + /// only the weights its record has in effect. + /// + /// The change applies from the next receipt this steward seals. + /// `precision` is the precision the new weights run at, `plan` the plan + /// the change falls under ([`plan_changes`](Self::plan_changes)) or `None` + /// for a change no plan foresaw, and `validation_result` the digest of + /// what its validation produced. A change outside its plan is refused, and + /// the backend keeps the weights it had. + pub fn record_weights( + &mut self, + engine: Box, + precision: Option, + plan: Option, + validation_result: Option, + reason: impl Into, + ) -> Result { + let backend = engine.backend_id(); + let spec = self.router.spec(&backend).ok_or_else(|| MeshError::NoEngine(backend.0.clone()))?; + let (recipe, fidelity) = (spec.recipe.clone(), spec.fidelity); + let to = engine + .weights_digest() + .ok_or_else(|| MeshError::ChangeRecord("an engine that names no weights cannot be recorded".into()))?; + let mut log = self.changes.get(&backend.0).cloned().unwrap_or_else(|| ChangeLog::new(backend.clone())); + let body = ChangeBody::Weights(WeightsChange { + from: log.current().and_then(ChangeEntry::weights).map(|w| w.to), + to, + precision, + fidelity, + plan, + validation_result, + effective_from: self.log.next_sequence(), + reason: reason.into(), + }); + let id = log + .append(recipe, (self.clock)(), body, &self.change_key, self.log.steward()) + .map_err(|e| MeshError::ChangeRecord(e.to_string()))?; + self.changes.insert(backend.0.clone(), log); + self.engines.insert(backend.0, engine); + Ok(id) + } + + /// Record, in a registered backend's change record, a plan for the + /// changes its steward foresees (SPEC §6.5), and return the plan's id, + /// which a change made under it names. A plan covers only the changes + /// recorded after it. + /// + /// A backend whose record holds a plan and no weights yet serves nothing + /// until its weights are recorded. + pub fn plan_changes(&mut self, backend: &str, plan: ChangePlan) -> Result { + let id = BackendId::new(backend); + let recipe = self.router.spec(&id).ok_or_else(|| MeshError::NoEngine(backend.into()))?.recipe.clone(); + let mut log = self.changes.get(backend).cloned().unwrap_or_else(|| ChangeLog::new(id)); + let plan_id = log + .append(recipe, (self.clock)(), ChangeBody::Plan(plan), &self.change_key, self.log.steward()) + .map_err(|e| MeshError::ChangeRecord(e.to_string()))?; + self.changes.insert(backend.into(), log); + Ok(plan_id) + } + + /// Hold a registered backend to the change record `sealed`, as the + /// steward persisted it from [`change_log`](Self::change_log): how a + /// restarted steward picks its records back up. Call it after + /// [`with_journal`](Self::with_journal), so the receipt log has its + /// position back. + /// + /// Every entry must be sealed under this steward's key and the record must + /// hold (SPEC §6.5). Its weights in effect must be the ones the backend's + /// engine runs, and must apply from a receipt the log has reached. + pub fn govern(&mut self, backend: &str, sealed: Vec<(ChangeEntry, Vec)>) -> Result<(), MeshError> { + let record = |reason: &str| MeshError::ChangeRecord(reason.into()); + let log = ChangeLog::from_sealed(BackendId::new(backend), sealed, self.log.steward(), &self.log.verifying_key()) + .map_err(|e| MeshError::ChangeRecord(e.to_string()))?; + let current = log.current().and_then(ChangeEntry::weights).ok_or_else(|| record("it records no weights"))?; + if current.effective_from > self.log.next_sequence() { + return Err(record("its weights apply from a receipt the log has not reached")); + } + let running = self.engines.get(backend).ok_or_else(|| MeshError::NoEngine(backend.into()))?.weights_digest(); + if running != Some(current.to) { + return Err(MeshError::WeightsUnrecorded { + backend: backend.into(), + running: running.map_or_else(|| "none".into(), |d| d.to_hex()), + recorded: current.to.to_hex(), + }); + } + self.changes.insert(backend.into(), log); + Ok(()) + } + + /// A backend's change record (SPEC §6.5), if it has one: what the steward + /// persists, and what it discloses beside the receipts it covers. + pub fn change_log(&self, backend: &str) -> Option<&ChangeLog> { + self.changes.get(backend) + } + /// The steward identity receipts are sealed under. pub fn steward(&self) -> &str { self.log.steward() @@ -474,6 +592,20 @@ impl<'m> Mesh<'m> { .ok_or_else(|| MeshError::NoRoute(route.justification.clone()))?; let model = self.engines.get(&backend.0).and_then(|e| e.weights_digest()); + // 1a. A backend with a change record serves only the weights the + // record has in effect (SPEC §6.5). Weights changed any other way + // are refused here, before any grant is decided or work runs. + if let Some(log) = self.changes.get(&backend.0) { + let recorded = log.current().and_then(ChangeEntry::weights).map(|w| w.to); + if recorded.is_none() || model != recorded { + return Err(MeshError::WeightsUnrecorded { + backend: backend.0.clone(), + running: model.map_or_else(|| "none".into(), |d| d.to_hex()), + recorded: recorded.map_or_else(|| "none".into(), |d| d.to_hex()), + }); + } + } + // 2. Authorise, and record the decision, before any work runs: a // decision recorded after the work would lose its debit if the // steward stopped in between (JCP §4.11). @@ -528,13 +660,17 @@ impl<'m> Mesh<'m> { }; // 3b. The engine must have run the weights it named before serving: - // those are the ones the grant was decided over. - if let (Some(named), Some(reported)) = (model, served.outcome.weights_digest) { - if named != reported { + // those are the ones the grant was decided over. A backend with a + // change record must also report them, or its receipt would name + // no weights for the record to cover. + let reported = served.outcome.weights_digest; + let unreported = reported.is_none() && self.changes.contains_key(&backend.0); + if let Some(named) = model { + if reported.is_some_and(|r| r != named) || unreported { return Err(MeshError::ModelMismatch { backend: backend.0.clone(), named: named.to_hex(), - reported: reported.to_hex(), + reported: reported.map_or_else(|| "none".into(), |d| d.to_hex()), }); } } @@ -1857,4 +1993,171 @@ mod tests { let again = mesh_with(&meter); assert!(resp.receipt.opens(&again.output_opening(&req.id()), &payload_digest(&[]), &resp.tokens)); } + + // ── the weights, changed only on the record (SPEC §6.5) ───────────────── + + /// The reference engine for `tiny`, naming the weights numbered `n`. + fn weighed(n: u8) -> Box { + let digest = ContentId::of_bytes(&[b"tiny weights ".as_slice(), &[n]].concat()); + Box::new(tiny().reporting_model_identity(digest, "int8")) + } + + fn tiny_spec() -> BackendSpec { + BackendSpec::new("tiny", "ssm-1b", Fidelity::F1, 50_000) + } + + /// Once a backend has a change record it serves only the weights the + /// record has in effect. Weights swapped in without a record are refused + /// before any work, and each receipt is covered by the change in effect + /// when it was sealed. + #[test] + fn a_recorded_backend_serves_only_the_weights_its_record_has_in_effect() { + let meter = SteppingMeter::new("test-rapl", Provenance::HwShunt, 1_000); + let mut mesh = mesh_with(&meter); + let req = request(16, Fidelity::F1, 5_000_000); + mesh.record_weights(weighed(1), Some("int8".into()), None, None, "the weights it starts on").unwrap(); + let first = mesh.serve(&req, &[], None).unwrap(); + + mesh.register(tiny_spec(), weighed(2)); + let (head, before) = (mesh.head(), meter.snapshot().uj); + assert!(matches!(mesh.serve(&req, &[], None), Err(MeshError::WeightsUnrecorded { .. }))); + assert_eq!(meter.snapshot().uj, before + 1_000, "refused before the meter was read: no work ran"); + assert_eq!(mesh.head(), head, "and nothing was sealed"); + + mesh.record_weights(weighed(2), Some("int8".into()), None, None, "retrained").unwrap(); + let second = mesh.serve(&req, &[], None).unwrap(); + let log = mesh.change_log("tiny").unwrap(); + assert_eq!(log.covers(&first.receipt).unwrap().seq, 0); + assert_eq!(log.covers(&second.receipt).unwrap().seq, 1); + assert_eq!(log.current().unwrap().weights().unwrap().from, first.receipt.weights_digest); + + // An engine that names no weights is refused under a record, and + // cannot be recorded. + mesh.register(tiny_spec(), Box::new(tiny())); + assert!(matches!(mesh.serve(&req, &[], None), Err(MeshError::WeightsUnrecorded { .. }))); + assert!(mesh.record_weights(Box::new(tiny()), None, None, None, "no weights").is_err()); + + // A record that holds only a plan has no weights in effect, so it + // serves nothing, not even an engine that names none. + let mut planned = mesh_with(&meter); + planned.plan_changes("tiny", plan_for_int8()).unwrap(); + assert!(matches!(planned.serve(&req, &[], None), Err(MeshError::WeightsUnrecorded { .. }))); + } + + /// An engine that names its weights before serving and reports none after + /// would seal a receipt naming no weights, which no record covers. + #[test] + fn a_recorded_backend_must_report_its_weights() { + struct Forgetting(Box); + impl ServingEngine for Forgetting { + fn backend_id(&self) -> BackendId { + self.0.backend_id() + } + fn recipe_id(&self) -> String { + self.0.recipe_id() + } + fn fidelity(&self) -> Fidelity { + self.0.fidelity() + } + fn est_uj_per_token(&self) -> u64 { + self.0.est_uj_per_token() + } + fn weights_digest(&self) -> Option { + self.0.weights_digest() + } + fn serve(&mut self, req: &ServeRequest, prompt: &[u32]) -> mesh_engine::Served { + let mut s = self.0.serve(req, prompt); + s.outcome.weights_digest = None; + s + } + } + let meter = SteppingMeter::new("test-rapl", Provenance::HwShunt, 1_000); + let req = request(16, Fidelity::F1, 5_000_000); + let mut unrecorded = mesh_with(&meter); + unrecorded.register(tiny_spec(), Box::new(Forgetting(weighed(1)))); + unrecorded.serve(&req, &[], None).unwrap(); + + let mut recorded = mesh_with(&meter); + recorded.record_weights(Box::new(Forgetting(weighed(1))), None, None, None, "forgets").unwrap(); + let head = recorded.head(); + assert!(matches!(recorded.serve(&req, &[], None), Err(MeshError::ModelMismatch { .. }))); + assert_eq!(recorded.head(), head, "nothing was sealed"); + } + + fn plan_for_int8() -> ChangePlan { + ChangePlan { + precisions: vec!["int8".into()], + fidelity_floor: Fidelity::F1, + validation: "the mesh/1 conformance pack".into(), + foresees: "requantization of the same checkpoint".into(), + risk_assessment: None, + } + } + + /// A change made under a plan is held to it. One outside it is refused, + /// and the backend keeps serving the weights it had. + #[test] + fn a_planned_weights_change_is_held_to_its_plan() { + let meter = SteppingMeter::new("test-rapl", Provenance::HwShunt, 1_000); + let mut mesh = mesh_with(&meter); + let req = request(16, Fidelity::F1, 5_000_000); + mesh.record_weights(weighed(1), Some("int8".into()), None, None, "the weights it starts on").unwrap(); + let plan = mesh.plan_changes("tiny", plan_for_int8()).unwrap(); + + let outside = mesh.record_weights(weighed(2), Some("int8".into()), Some(plan), None, "requantized"); + assert!(matches!(outside, Err(MeshError::ChangeRecord(r)) if r.contains("no validation result"))); + let kept = mesh.serve(&req, &[], None).unwrap(); + assert_eq!(mesh.change_log("tiny").unwrap().covers(&kept.receipt).unwrap().seq, 0); + + let passed = Some(ContentId::of_bytes(b"conformance result")); + mesh.record_weights(weighed(2), Some("int8".into()), Some(plan), passed, "requantized").unwrap(); + let resp = mesh.serve(&req, &[], None).unwrap(); + let log = mesh.change_log("tiny").unwrap(); + let change = log.covers(&resp.receipt).unwrap().weights().unwrap(); + assert_eq!(log.plan_of(change).unwrap().foresees, "requantization of the same checkpoint"); + assert_eq!(change.validation_result, passed); + } + + /// A restarted steward picks its change record back up from the entries + /// it persisted, if they hold and its engine runs the weights they have in + /// effect, and not otherwise. + #[test] + fn a_restarted_steward_governs_from_its_persisted_record() { + let meter = SteppingMeter::new("test-rapl", Provenance::HwShunt, 1_000); + let req = request(16, Fidelity::F1, 5_000_000); + let mut mesh = mesh_with(&meter); + mesh.record_weights(weighed(1), Some("int8".into()), None, None, "the weights it starts on").unwrap(); + let sealed = mesh.change_log("tiny").unwrap().entries().to_vec(); + + let restarted = |engine: Box| { + let mut m = mesh_with(&meter); + m.register(tiny_spec(), engine); + m + }; + let mut again = restarted(weighed(1)); + again.govern("tiny", sealed.clone()).unwrap(); + again.serve(&req, &[], None).unwrap(); + again.register(tiny_spec(), weighed(2)); + assert!(matches!(again.serve(&req, &[], None), Err(MeshError::WeightsUnrecorded { .. })), "still held to it"); + + assert!(matches!(restarted(weighed(2)).govern("tiny", sealed.clone()), Err(MeshError::WeightsUnrecorded { .. }))); + // A change that applies from a receipt the restarted log has not + // reached is not one it can hold a backend to. + mesh.serve(&req, &[], None).unwrap(); + mesh.record_weights(weighed(2), Some("int8".into()), None, None, "retrained").unwrap(); + let ahead = mesh.change_log("tiny").unwrap().entries().to_vec(); + let refused = restarted(weighed(2)).govern("tiny", ahead); + assert!(matches!(refused, Err(MeshError::ChangeRecord(r)) if r.contains("not reached"))); + let mut tampered = sealed.clone(); + tampered[0].0.made_at += 1; + assert!(matches!(restarted(weighed(1)).govern("tiny", tampered), Err(MeshError::ChangeRecord(_)))); + let (other_sk, other) = (SigningKey::from_bytes(&[42u8; 32]), "another-steward"); + let mut foreign = ChangeLog::new(BackendId::new("tiny")); + let body = sealed[0].0.body.clone(); + foreign.append("ssm-1b", 1, body, &other_sk, other).unwrap(); + assert!(matches!( + restarted(weighed(1)).govern("tiny", foreign.entries().to_vec()), + Err(MeshError::ChangeRecord(_)) + )); + } } diff --git a/mesh/mesh-rs/crates/mesh-receipt/src/change.rs b/mesh/mesh-rs/crates/mesh-receipt/src/change.rs new file mode 100644 index 000000000..c3c98b795 --- /dev/null +++ b/mesh/mesh-rs/crates/mesh-receipt/src/change.rs @@ -0,0 +1,640 @@ +//! The weights change record (SPEC §6.5). +//! +//! A receipt's `weights_digest` names the weights that answered. Nothing +//! recorded when a backend's weights changed, from what, or whether the change +//! was one its steward had planned for. A [`ChangeLog`] is that record, one per +//! backend: a hash-chained sequence of entries, each sealed in the JCR-1 +//! envelope receipts use, under the same steward key. An entry is one of two +//! kinds: +//! +//! - A [`ChangePlan`] records, before the changes it covers, which changes the +//! steward foresees for the backend: the precisions it may run at, the +//! fidelity it must keep, and the validation each change must pass. +//! - A [`WeightsChange`] records one change: the weights before and after, the +//! plan it falls under or none, the digest of its validation result, and the +//! position in the steward's receipt log from which it applies. +//! +//! A plan covers only the changes sealed after it, so a change recorded as +//! planned was foreseen before it was made, by the chain's own order. A change +//! no plan covers is recorded too, as unplanned: the log records what changed, +//! and does not judge it. +//! +//! A change applies from a receipt-log position rather than from a time, so +//! [`ChangeLog::covers`] can say which recorded change put a receipt's weights +//! in effect without trusting a clock. Each entry also carries the time its +//! steward made it. + +use ed25519_dalek::SigningKey; +use jcp_schema::ContentId; +use mesh_schema::{BackendId, Fidelity}; +use serde::{Deserialize, Serialize}; +use serde_json::json; + +use crate::{jcr1, MeshReceipt}; + +/// The changes a steward foresees for a backend, recorded before them. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ChangePlan { + /// The precisions a planned change may run at. Empty: the plan does not + /// constrain precision. + pub precisions: Vec, + /// The fidelity the backend must still be rated at after a planned change. + pub fidelity_floor: Fidelity, + /// The validation every planned change must pass, as the steward names it + /// (a conformance pack, an evaluation suite). A change under the plan + /// carries the digest of its result. + pub validation: String, + /// What the plan foresees, in the steward's words. + pub foresees: String, + /// The digest of the risk assessment the plan rests on, where there is one. + pub risk_assessment: Option, +} + +/// One change of a backend's weights. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WeightsChange { + /// The weights before the change; `None` for the backend's first record. + pub from: Option, + /// The weights after it. + pub to: ContentId, + /// The precision the new weights run at, where the steward names it. + pub precision: Option, + /// The backend's fidelity rating after the change. + pub fidelity: Fidelity, + /// The plan entry this change falls under; `None` for a change no plan + /// foresaw. + pub plan: Option, + /// The digest of the change's validation result. + pub validation_result: Option, + /// The first sequence in the steward's receipt log served on the new + /// weights. + pub effective_from: u64, + /// Why the weights changed, in the steward's words. + pub reason: String, +} + +/// What an entry records. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ChangeBody { + Plan(ChangePlan), + Weights(WeightsChange), +} + +/// One sealed entry of a backend's change log. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ChangeEntry { + pub backend: BackendId, + pub recipe: String, + /// The entry's position in its log. + pub seq: u64, + /// The content id of the envelope before it; `None` for the first. + pub prev: Option, + /// Unix seconds at which the steward made the entry. + pub made_at: u64, + pub body: ChangeBody, +} + +impl ChangeEntry { + /// The weights change this entry records, if it records one. + pub fn weights(&self) -> Option<&WeightsChange> { + match &self.body { + ChangeBody::Weights(w) => Some(w), + ChangeBody::Plan(_) => None, + } + } + + /// The plan this entry records, if it records one. + pub fn plan(&self) -> Option<&ChangePlan> { + match &self.body { + ChangeBody::Plan(p) => Some(p), + ChangeBody::Weights(_) => None, + } + } + + /// The JCR-1 wire projection. Optional fields are omitted when absent. + pub fn to_jcr1_wire(&self, steward: &str) -> serde_json::Value { + let mut obj = serde_json::Map::new(); + obj.insert("mesh_change".into(), json!("mesh/1")); + obj.insert("backend".into(), json!(self.backend.0)); + obj.insert("recipe".into(), json!(self.recipe)); + obj.insert("seq".into(), json!(self.seq)); + if let Some(p) = &self.prev { + obj.insert("prev".into(), json!(p.to_hex())); + } + obj.insert("made_at".into(), json!(self.made_at)); + match &self.body { + ChangeBody::Plan(p) => { + obj.insert("kind".into(), json!("plan")); + obj.insert("precisions".into(), json!(p.precisions)); + obj.insert("fidelity_floor".into(), json!(p.fidelity_floor.0)); + obj.insert("validation".into(), json!(p.validation)); + obj.insert("foresees".into(), json!(p.foresees)); + if let Some(r) = &p.risk_assessment { + obj.insert("risk_assessment".into(), json!(r.to_hex())); + } + } + ChangeBody::Weights(w) => { + obj.insert("kind".into(), json!("weights")); + if let Some(f) = &w.from { + obj.insert("from".into(), json!(f.to_hex())); + } + obj.insert("to".into(), json!(w.to.to_hex())); + if let Some(p) = &w.precision { + obj.insert("precision".into(), json!(p)); + } + obj.insert("fidelity".into(), json!(w.fidelity.0)); + if let Some(p) = &w.plan { + obj.insert("plan".into(), json!(p.to_hex())); + } + if let Some(v) = &w.validation_result { + obj.insert("validation_result".into(), json!(v.to_hex())); + } + obj.insert("effective_from".into(), json!(w.effective_from)); + obj.insert("reason".into(), json!(w.reason)); + } + } + obj.insert("steward".into(), json!(steward)); + serde_json::Value::Object(obj) + } + + /// The canonical signed-core bytes (JCR-1 det-CBOR). + pub fn signed_core(&self, steward: &str) -> Result, jcr1::Jcr1Error> { + jcr1::signed_core_bytes(&self.to_jcr1_wire(steward)) + } +} + +/// Why a change log, or an entry offered to one, does not hold, or why it +/// does not cover a receipt. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum ChangeError { + /// An entry breaks the log's rules (SPEC §6.5). + #[error("change entry {seq}: {reason}")] + Entry { seq: u64, reason: &'static str }, + /// An entry's envelope does not verify, or seals another entry. + #[error("change entry {seq}: envelope: {detail}")] + Envelope { seq: u64, detail: String }, + /// The log says nothing that puts the receipt's weights in effect. + #[error("not covered by the change log: {0}")] + NotCovered(&'static str), +} + +/// A backend's weights change record (SPEC §6.5). +/// +/// Every entry is checked as it joins, by [`append`](Self::append) when the +/// steward makes it and by [`from_sealed`](Self::from_sealed) when a party +/// reads a log it was handed, so a log that exists holds. +#[derive(Clone, Debug)] +pub struct ChangeLog { + backend: BackendId, + entries: Vec<(ChangeEntry, Vec)>, +} + +impl ChangeLog { + /// An empty log for `backend`. + pub fn new(backend: BackendId) -> Self { + Self { backend, entries: Vec::new() } + } + + /// Read a log from its sealed entries, checking each envelope against the + /// steward's key and each entry against the ones before it. + pub fn from_sealed( + backend: BackendId, + sealed: Vec<(ChangeEntry, Vec)>, + steward: &str, + verifying_key: &[u8; 32], + ) -> Result { + let mut log = Self::new(backend); + for (entry, envelope) in sealed { + let envelope_err = |detail: String| ChangeError::Envelope { seq: entry.seq, detail }; + let payload = jcr1::verify_cose_sign1(&envelope, steward, verifying_key).map_err(|e| envelope_err(e.to_string()))?; + let core = entry.signed_core(steward).map_err(|e| envelope_err(e.to_string()))?; + if payload != core { + return Err(envelope_err("it seals another entry".into())); + } + log.check_next(&entry)?; + log.entries.push((entry, envelope)); + } + Ok(log) + } + + /// Make, check and seal the next entry, returning its id. + pub fn append( + &mut self, + recipe: impl Into, + made_at: u64, + body: ChangeBody, + key: &SigningKey, + steward: &str, + ) -> Result { + let entry = ChangeEntry { + backend: self.backend.clone(), + recipe: recipe.into(), + seq: self.entries.len() as u64, + prev: self.head(), + made_at, + body, + }; + self.check_next(&entry)?; + let core = entry + .signed_core(steward) + .map_err(|e| ChangeError::Envelope { seq: entry.seq, detail: e.to_string() })?; + let envelope = jcr1::cose_sign1_envelope(&core, key, steward); + let id = ContentId::of_bytes(&envelope); + self.entries.push((entry, envelope)); + Ok(id) + } + + /// The backend this log records. + pub fn backend(&self) -> &BackendId { + &self.backend + } + + /// Every entry with its envelope, in order: what a steward persists, and + /// what [`from_sealed`](Self::from_sealed) reads back. + pub fn entries(&self) -> &[(ChangeEntry, Vec)] { + &self.entries + } + + /// The content id of the last entry's envelope. + pub fn head(&self) -> Option { + self.entries.last().map(|(_, env)| ContentId::of_bytes(env)) + } + + /// The last weights change: the weights the backend runs now. + pub fn current(&self) -> Option<&ChangeEntry> { + self.entries.iter().rev().map(|(e, _)| e).find(|e| e.weights().is_some()) + } + + /// The weights change in effect at `sequence` of the steward's receipt + /// log: the last one that applies from that position or before it. + pub fn in_effect_at(&self, sequence: u64) -> Option<&ChangeEntry> { + self.entries + .iter() + .rev() + .map(|(e, _)| e) + .find(|e| e.weights().is_some_and(|w| w.effective_from <= sequence)) + } + + /// The plan `change` falls under, or `None` for an unplanned change. + pub fn plan_of(&self, change: &WeightsChange) -> Option<&ChangePlan> { + change.plan.and_then(|id| self.plan_by_id(&id)) + } + + /// Which recorded change put `receipt`'s weights in effect. + /// + /// The receipt must be one this steward sealed, which its envelope shows + /// ([`MeshReceipt::verify_envelope`]): its position is in the same + /// receipt log the changes apply from. + pub fn covers(&self, receipt: &MeshReceipt) -> Result<&ChangeEntry, ChangeError> { + let not = ChangeError::NotCovered; + if receipt.backend != self.backend { + return Err(not("the receipt is for another backend")); + } + let Some(weights) = receipt.weights_digest else { + return Err(not("the receipt names no weights")); + }; + let entry = self + .in_effect_at(receipt.sequence) + .ok_or(not("no recorded change applies at the receipt's position"))?; + let change = entry.weights().expect("in_effect_at yields weights changes"); + if change.to != weights { + return Err(not("the receipt's weights are not the ones recorded in effect")); + } + if entry.recipe != receipt.recipe { + return Err(not("the receipt's recipe is not the one recorded in effect")); + } + if let (Some(recorded), Some(sealed)) = (&change.precision, &receipt.precision) { + if recorded != sealed { + return Err(not("the receipt's precision is not the one recorded in effect")); + } + } + Ok(entry) + } + + fn plan_by_id(&self, id: &ContentId) -> Option<&ChangePlan> { + self.entries + .iter() + .find(|(_, env)| ContentId::of_bytes(env) == *id) + .and_then(|(e, _)| e.plan()) + } + + /// The rules an entry must meet to follow the ones already in the log. + fn check_next(&self, entry: &ChangeEntry) -> Result<(), ChangeError> { + let bad = |reason: &'static str| Err(ChangeError::Entry { seq: entry.seq, reason }); + if entry.backend != self.backend { + return bad("it is for another backend"); + } + if entry.seq != self.entries.len() as u64 || entry.prev != self.head() { + return bad("it does not chain to the entry before it"); + } + if self.entries.last().is_some_and(|(last, _)| entry.made_at < last.made_at) { + return bad("it was made before the entry it follows"); + } + let ChangeBody::Weights(w) = &entry.body else { + return Ok(()); + }; + let current = self.current(); + if w.from != current.and_then(|e| e.weights()).map(|c| c.to) { + return bad("it does not change from the weights in effect"); + } + if w.from == Some(w.to) { + return bad("it changes nothing"); + } + if current.is_some_and(|e| e.weights().is_some_and(|c| w.effective_from < c.effective_from)) { + return bad("it applies before the change it follows"); + } + let Some(plan_id) = &w.plan else { + return Ok(()); + }; + let Some((plan_entry, _)) = self + .entries + .iter() + .find(|(e, env)| e.plan().is_some() && ContentId::of_bytes(env) == *plan_id) + else { + return bad("it names no plan recorded before it"); + }; + let plan = plan_entry.plan().expect("found as a plan"); + if plan_entry.recipe != entry.recipe { + return bad("outside its plan: another recipe"); + } + if !plan.precisions.is_empty() && !w.precision.as_ref().is_some_and(|p| plan.precisions.contains(p)) { + return bad("outside its plan: a precision the plan does not foresee"); + } + if w.fidelity < plan.fidelity_floor { + return bad("outside its plan: below the plan's fidelity floor"); + } + if w.validation_result.is_none() { + return bad("outside its plan: no validation result"); + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use mesh_schema::{Energy, ServeOutcome}; + + const STEWARD: &str = "steward-1"; + + fn key() -> SigningKey { + SigningKey::from_bytes(&[5u8; 32]) + } + + fn vk() -> [u8; 32] { + key().verifying_key().to_bytes() + } + + fn weights(n: u8) -> ContentId { + ContentId::of_bytes(&[b"weights ".as_slice(), &[n]].concat()) + } + + fn plan() -> ChangeBody { + ChangeBody::Plan(ChangePlan { + precisions: vec!["int8".into(), "bf16".into()], + fidelity_floor: Fidelity::F2, + validation: "mesh/1 conformance pack + held-out suite".into(), + foresees: "requantization of the same checkpoint".into(), + risk_assessment: Some(ContentId::of_bytes(b"risk assessment v1")), + }) + } + + fn change(from: Option, to: u8, plan: Option, effective_from: u64) -> ChangeBody { + ChangeBody::Weights(WeightsChange { + from: from.map(weights), + to: weights(to), + precision: Some("int8".into()), + fidelity: Fidelity::F2, + plan, + validation_result: plan.map(|_| ContentId::of_bytes(b"validation passed")), + effective_from, + reason: "requantized".into(), + }) + } + + /// A plan, the backend's first weights, and a change under the plan. + fn three() -> (ChangeLog, ContentId) { + let mut log = ChangeLog::new(BackendId::new("tiny")); + let p = log.append("ssm-1b", 100, plan(), &key(), STEWARD).unwrap(); + log.append("ssm-1b", 110, change(None, 1, None, 0), &key(), STEWARD).unwrap(); + log.append("ssm-1b", 120, change(Some(1), 2, Some(p), 7), &key(), STEWARD).unwrap(); + (log, p) + } + + fn receipt(sequence: u64, weights_digest: Option) -> MeshReceipt { + MeshReceipt { + request_id: ContentId::of_bytes(b"request"), + grant_id: None, + backend: BackendId::new("tiny"), + recipe: "ssm-1b".into(), + weights_digest, + precision: Some("int8".into()), + route_justification: "test".into(), + outcome: ServeOutcome { tokens_in: 1, tokens_out: 1, tokens_accepted: 1, weights_digest, ..Default::default() }, + fidelity_floor: Fidelity::F1, + fidelity_advertised: None, + fidelity_delivered: None, + fidelity_met: true, + energy: Energy { estimated_uj: 1, measured_uj: None, provenance: crate::Provenance::Estimator }, + grant_breach: None, + output_commitment: None, + sequence, + prev: None, + nonce: [0; 8], + } + } + + /// A log made by its steward reads back, whole, from its sealed entries, + /// and says which plan a change fell under. + #[test] + fn a_change_log_reads_back_from_its_sealed_entries() { + let (log, p) = three(); + let read = ChangeLog::from_sealed(log.backend().clone(), log.entries().to_vec(), STEWARD, &vk()).unwrap(); + assert_eq!(read.head(), log.head()); + let current = read.current().unwrap().weights().unwrap(); + assert_eq!(current.to, weights(2)); + assert_eq!(current.plan, Some(p)); + assert_eq!(read.plan_of(current).unwrap().validation, "mesh/1 conformance pack + held-out suite"); + let first = read.in_effect_at(0).unwrap().weights().unwrap(); + assert_eq!(read.plan_of(first), None, "the first weights fell under no plan"); + } + + /// Altered, reordered, truncated at the front or sealed by another key, + /// a log does not read. + #[test] + fn a_tampered_log_does_not_read() { + let (log, _) = three(); + let read = |sealed: Vec<(ChangeEntry, Vec)>, vk: &[u8; 32]| { + ChangeLog::from_sealed(BackendId::new("tiny"), sealed, STEWARD, vk) + }; + + let mut altered = log.entries().to_vec(); + if let ChangeBody::Weights(w) = &mut altered[2].0.body { + w.reason = "something else".into(); + } + assert!(matches!(read(altered, &vk()), Err(ChangeError::Envelope { seq: 2, .. }))); + + let mut swapped = log.entries().to_vec(); + swapped.swap(1, 2); + assert!(matches!(read(swapped, &vk()), Err(ChangeError::Entry { seq: 2, .. }))); + + let dropped = log.entries()[1..].to_vec(); + assert!(matches!(read(dropped, &vk()), Err(ChangeError::Entry { seq: 1, .. }))); + + let other = SigningKey::from_bytes(&[6u8; 32]).verifying_key().to_bytes(); + assert!(matches!(read(log.entries().to_vec(), &other), Err(ChangeError::Envelope { seq: 0, .. }))); + } + + /// Each chaining rule holds by itself. An entry sealed by the steward that + /// breaks only one of them (its backend, its position, or its link to the + /// entry before it) does not join the log. + #[test] + fn an_entry_joins_only_where_it_chains() { + let (log, _) = three(); + let sealed = |e: ChangeEntry| { + let env = jcr1::cose_sign1_envelope(&e.signed_core(STEWARD).unwrap(), &key(), STEWARD); + let mut all = log.entries().to_vec(); + all.push((e, env)); + ChangeLog::from_sealed(BackendId::new("tiny"), all, STEWARD, &vk()) + }; + let next = ChangeEntry { + backend: BackendId::new("tiny"), + recipe: "ssm-1b".into(), + seq: 3, + prev: log.head(), + made_at: 130, + body: change(Some(2), 3, None, 9), + }; + let reason = |r: Result| match r { + Err(ChangeError::Entry { reason, .. }) => reason, + other => panic!("joined: {other:?}"), + }; + assert_eq!(reason(sealed(ChangeEntry { backend: BackendId::new("frontier"), ..next.clone() })), "it is for another backend"); + assert_eq!(reason(sealed(ChangeEntry { seq: 4, ..next.clone() })), "it does not chain to the entry before it"); + let elsewhere = Some(ContentId::of_bytes(b"another envelope")); + assert_eq!(reason(sealed(ChangeEntry { prev: elsewhere, ..next.clone() })), "it does not chain to the entry before it"); + sealed(next).unwrap(); + } + + /// A change starts from the weights in effect, changes something, and + /// applies no earlier than the change before it; entries keep time order. + #[test] + fn a_change_follows_the_weights_in_effect() { + let (mut log, _) = three(); + let reason = |r: Result| match r { + Err(ChangeError::Entry { reason, .. }) => reason, + other => panic!("accepted: {other:?}"), + }; + assert_eq!( + reason(log.append("ssm-1b", 130, change(Some(1), 3, None, 9), &key(), STEWARD)), + "it does not change from the weights in effect" + ); + assert_eq!(reason(log.append("ssm-1b", 130, change(Some(2), 2, None, 9), &key(), STEWARD)), "it changes nothing"); + assert_eq!( + reason(log.append("ssm-1b", 130, change(Some(2), 3, None, 6), &key(), STEWARD)), + "it applies before the change it follows" + ); + assert_eq!( + reason(log.append("ssm-1b", 119, change(Some(2), 3, None, 9), &key(), STEWARD)), + "it was made before the entry it follows" + ); + // Unplanned is recorded, as unplanned. + log.append("ssm-1b", 130, change(Some(2), 3, None, 9), &key(), STEWARD).unwrap(); + assert_eq!(log.plan_of(log.current().unwrap().weights().unwrap()), None); + } + + /// A change recorded under a plan is one the plan foresaw: recorded + /// before it, for its recipe, at a precision it lists, at or above its + /// fidelity floor, with a validation result. + #[test] + fn a_planned_change_stays_inside_its_plan() { + let (mut log, p) = three(); + let reason = |r: Result| match r { + Err(ChangeError::Entry { reason, .. }) => reason, + other => panic!("accepted: {other:?}"), + }; + let with = |f: &dyn Fn(&mut WeightsChange)| { + let mut body = change(Some(2), 3, Some(p), 9); + if let ChangeBody::Weights(w) = &mut body { + f(w); + } + body + }; + assert_eq!( + reason(log.append("ssm-1b", 130, with(&|w| w.precision = Some("fp16".into())), &key(), STEWARD)), + "outside its plan: a precision the plan does not foresee" + ); + assert_eq!( + reason(log.append("ssm-1b", 130, with(&|w| w.precision = None), &key(), STEWARD)), + "outside its plan: a precision the plan does not foresee" + ); + assert_eq!( + reason(log.append("ssm-1b", 130, with(&|w| w.fidelity = Fidelity::F1), &key(), STEWARD)), + "outside its plan: below the plan's fidelity floor" + ); + assert_eq!( + reason(log.append("ssm-1b", 130, with(&|w| w.validation_result = None), &key(), STEWARD)), + "outside its plan: no validation result" + ); + assert_eq!(reason(log.append("ssm-2b", 130, with(&|_| ()), &key(), STEWARD)), "outside its plan: another recipe"); + // A weights entry's id is not a plan, and neither is an unknown id. + let not_a_plan = log.head().unwrap(); + assert_eq!( + reason(log.append("ssm-1b", 130, with(&|w| w.plan = Some(not_a_plan)), &key(), STEWARD)), + "it names no plan recorded before it" + ); + log.append("ssm-1b", 130, with(&|_| ()), &key(), STEWARD).unwrap(); + } + + /// A plan covers only the changes sealed after it. One sealed later does + /// not reach back over an earlier change, and a change can name only a + /// plan already in its own log. + #[test] + fn a_plan_covers_only_what_comes_after_it() { + let mut log = ChangeLog::new(BackendId::new("tiny")); + log.append("ssm-1b", 100, change(None, 1, None, 0), &key(), STEWARD).unwrap(); + let p = log.append("ssm-1b", 101, plan(), &key(), STEWARD).unwrap(); + let first = log.in_effect_at(0).unwrap().weights().unwrap(); + assert_eq!(log.plan_of(first), None, "a plan sealed later does not cover it"); + + // The same plan sealed in another log is another entry, with another id. + let mut other = ChangeLog::new(BackendId::new("tiny")); + let elsewhere = other.append("ssm-1b", 100, plan(), &key(), STEWARD).unwrap(); + assert_ne!(elsewhere, p); + let r = log.append("ssm-1b", 102, change(Some(1), 2, Some(elsewhere), 3), &key(), STEWARD); + assert!(matches!(r, Err(ChangeError::Entry { reason: "it names no plan recorded before it", .. }))); + log.append("ssm-1b", 102, change(Some(1), 2, Some(p), 3), &key(), STEWARD).unwrap(); + } + + /// A receipt is covered by the change in effect at its position, for its + /// backend, recipe, weights and precision, and by nothing else. + #[test] + fn a_receipt_is_covered_by_the_change_in_effect_at_its_position() { + let (log, _) = three(); + assert_eq!(log.covers(&receipt(0, Some(weights(1)))).unwrap().seq, 1); + assert_eq!(log.covers(&receipt(6, Some(weights(1)))).unwrap().seq, 1); + assert_eq!(log.covers(&receipt(7, Some(weights(2)))).unwrap().seq, 2); + + let not = |r: Result<&ChangeEntry, ChangeError>| match r { + Err(ChangeError::NotCovered(why)) => why, + other => panic!("covered: {other:?}"), + }; + assert!(not(log.covers(&receipt(7, Some(weights(1))))).contains("weights")); + assert!(not(log.covers(&receipt(6, Some(weights(2))))).contains("weights")); + assert!(not(log.covers(&receipt(7, None))).contains("names no weights")); + let mut other_backend = receipt(7, Some(weights(2))); + other_backend.backend = BackendId::new("frontier"); + assert!(not(log.covers(&other_backend)).contains("another backend")); + let mut other_recipe = receipt(7, Some(weights(2))); + other_recipe.recipe = "ssm-2b".into(); + assert!(not(log.covers(&other_recipe)).contains("recipe")); + let mut other_precision = receipt(7, Some(weights(2))); + other_precision.precision = Some("bf16".into()); + assert!(not(log.covers(&other_precision)).contains("precision")); + + let mut late = ChangeLog::new(BackendId::new("tiny")); + late.append("ssm-1b", 100, change(None, 1, None, 5), &key(), STEWARD).unwrap(); + assert!(not(late.covers(&receipt(4, Some(weights(1))))).contains("no recorded change")); + } +} diff --git a/mesh/mesh-rs/crates/mesh-receipt/src/lib.rs b/mesh/mesh-rs/crates/mesh-receipt/src/lib.rs index eb6f99b73..918e37e84 100644 --- a/mesh/mesh-rs/crates/mesh-receipt/src/lib.rs +++ b/mesh/mesh-rs/crates/mesh-receipt/src/lib.rs @@ -14,6 +14,7 @@ use mesh_schema::{BackendId, Energy, Fidelity, SchemaError, ServeOutcome}; use serde::{Deserialize, Serialize}; use serde_json::json; +pub mod change; pub mod jcr1; pub use mesh_schema::Provenance;