From b2447e4f9936a53c3b680b472f6eff02d379a145 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 10 Oct 2026 13:56:43 +0200 Subject: [PATCH 01/10] Hosted lane tests: pin DEL (0x7f) as a key byte that cannot ride a header Follow-up hosted-key-del-pin: a mutant allowing DEL survived the key-shape test. The case list gains an interior DEL. Builds: yes. Co-Authored-By: Claude Opus 5.5 --- crates/ghostlight/src/local_inference.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/ghostlight/src/local_inference.rs b/crates/ghostlight/src/local_inference.rs index 97bb460..fd7cdff 100644 --- a/crates/ghostlight/src/local_inference.rs +++ b/crates/ghostlight/src/local_inference.rs @@ -1996,7 +1996,7 @@ mod tests { #[test] fn hosted_binding_refuses_a_key_shape_that_cannot_ride_a_header() { let directory = tempfile::tempdir().unwrap(); - let cases: [(&str, &[u8]); 8] = [ + let cases: [(&str, &[u8]); 9] = [ ("interior newline", b"keycanary\nsecondline"), ("interior carriage return", b"keycanary\rsecondline"), ("interior tab", b"keycanary\tsecondline"), @@ -2005,6 +2005,7 @@ mod tests { ("trailing space before the newline", b"keycanary \n"), ("non-ascii", "keycanary-\u{e9}".as_bytes()), ("leading newline", b"\nkeycanary"), + ("interior DEL", b"keycanary\x7fsecondline"), ]; for (label, content) in cases { let path = write_key_file(&directory, content); From e28bf3fd559ab04ffcff885e1c3eb5ef612adfd5 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 10 Oct 2026 14:02:48 +0200 Subject: [PATCH 02/10] Verse call records: a recording port owns the run's spend, a replay port serves the stored responses ProviderCallRecord (ghostlight.verse_provider_call.v1) keyed (run id, sequence), CallRecordStore over an owned redb CultCache file, RecordingInferencePort (bound check before the call, real usage charged after, record written before the output returns) and ReplayInferencePort. InferenceFaultClass gains BudgetExhausted. Not yet compiled or run: this commit may not build. Co-Authored-By: Claude Opus 5.5 --- crates/ghostlight/src/controllers.rs | 11 + crates/ghostlight/src/lib.rs | 5 + crates/ghostlight/src/verse/calls.rs | 474 +++++++++++++++ crates/ghostlight/src/verse/calls/tests.rs | 652 +++++++++++++++++++++ crates/ghostlight/src/verse/mod.rs | 10 + 5 files changed, 1152 insertions(+) create mode 100644 crates/ghostlight/src/verse/calls.rs create mode 100644 crates/ghostlight/src/verse/calls/tests.rs create mode 100644 crates/ghostlight/src/verse/mod.rs diff --git a/crates/ghostlight/src/controllers.rs b/crates/ghostlight/src/controllers.rs index 6eccdca..b2fae5d 100644 --- a/crates/ghostlight/src/controllers.rs +++ b/crates/ghostlight/src/controllers.rs @@ -250,6 +250,9 @@ pub enum InferenceFaultClass { Status(u16), /// The endpoint's reply was not the declared shape. BadReply, + /// A recording port refused the call because it could pass the run's + /// token cap. Nothing was sent; the run stops and does not retry. + BudgetExhausted, Unclassified, } @@ -300,6 +303,14 @@ impl InferenceFault { Self::new(detail) } + /// The run's token cap would be passed by this call, so it was not made. + /// An integrity violation is the one disposition that stops the owning + /// flow without retrying, which is what a spent budget asks for. + pub(crate) fn budget_exhausted() -> Self { + Self::integrity_violation("the run's token cap would be passed by this call") + .classed(InferenceFaultClass::BudgetExhausted) + } + /// Names what kind of trouble this fault is. pub(super) fn classed(mut self, class: InferenceFaultClass) -> Self { self.class = class; diff --git a/crates/ghostlight/src/lib.rs b/crates/ghostlight/src/lib.rs index 0a24f3c..409a251 100644 --- a/crates/ghostlight/src/lib.rs +++ b/crates/ghostlight/src/lib.rs @@ -70,6 +70,7 @@ mod sdk_inference; mod table; mod tool_schema; mod vault; +mod verse; pub(crate) use action::ActionMismatch; pub(crate) use clock::Motion; @@ -130,6 +131,10 @@ pub use table::{ id_text, id_text_matches, table_view, }; pub use vault::{VaultError, VaultEvidenceSource}; +pub use verse::{ + CALL_RECORD_SCHEMA, CallRecordError, CallRecordStore, ProviderCallRecord, RecordingInferencePort, + ReplayInferencePort, RunId, +}; use patch::{EdgeRecord, EntityRecord, LedgerDelta, ResolvedOp, ResolvedPatch}; use chrono::{DateTime, Utc}; diff --git a/crates/ghostlight/src/verse/calls.rs b/crates/ghostlight/src/verse/calls.rs new file mode 100644 index 0000000..328417b --- /dev/null +++ b/crates/ghostlight/src/verse/calls.rs @@ -0,0 +1,474 @@ +//! Provider call records, the run's spend, and replay. +//! +//! `RecordingInferencePort` is the one owner of a run's spend. Before it +//! forwards a call it decides, from the request alone, whether the call could +//! pass the run's token cap; after the reply it charges what the call really +//! cost. Every reply is stored as a `ProviderCallRecord` before the output +//! returns to the lane, so `ReplayInferencePort` can serve the same run again +//! from the store with no provider behind it. +//! +//! Lanes and runners never count tokens. A request is held against the cap by +//! its bound: the serialized provider request's bytes (a token is at least one +//! byte, so bytes bound the prompt) plus the request's own output allowance. +//! The bound is exact for a port that makes one provider round per call (the +//! hosted and local lanes). A port that loops tools inside one call can spend +//! more than one round's bound; its settled usage is still charged, so the next +//! call is refused, but this decorator cannot stop the overshoot of that call. +//! +//! Charging rule, one path: a reply that reports usage is charged prompt plus +//! completion (cached prompt tokens are part of the prompt); a reply that +//! reports none is charged its bound; a fault is charged its bound, except a +//! connect fault, which never reached the provider and costs nothing. + +use crate::controllers::{ + InferenceEvent, InferenceFault, InferenceFaultClass, InferenceOutput, InferencePort, + InferencePurpose, InferenceRequest, PreparedInference, REQUEST_EXPIRY, TokenUsage, + ToolResultOracle, unix_ms, +}; +use async_trait::async_trait; +use chrono::Utc; +use cultcache_rs::{CacheBackingStore, CultCacheEnvelope, OwnedRedbMessagePackBackingStore}; +use serde::{Deserialize, Serialize}; +use std::{ + collections::{BTreeMap, VecDeque}, + path::Path, + sync::{Arc, Mutex}, +}; +use thiserror::Error; + +const CALL_RECORD_ROW: &str = "verse_provider_call.v1"; +/// The call record document's schema. A store holding any other row type or +/// schema is refused at open: this store is not migrated from, or shared with, +/// another document. +pub const CALL_RECORD_SCHEMA: &str = "ghostlight.verse_provider_call.v1"; + +#[derive(Clone, Copy, Debug, Error, PartialEq, Eq)] +pub enum CallRecordError { + #[error("a run id is 1 to 64 characters of letters, digits, '-' and '_'")] + InvalidRunId, + #[error("the call record store could not be read or written")] + Store, + #[error("the call record store holds a row that is not a canonical call record")] + Corrupt, + #[error("a call record's sequence is not the next in its run")] + OutOfSequence, + #[error("replay needs one run's records in dense sequence from 1")] + NotReplayable, +} + +/// A run's identity as a key segment. +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub struct RunId(String); + +impl RunId { + pub fn new(id: &str) -> Result { + let valid = !id.is_empty() + && id.len() <= 64 + && id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_'); + if valid { + Ok(Self(id.to_owned())) + } else { + Err(CallRecordError::InvalidRunId) + } + } + + pub fn as_str(&self) -> &str { + &self.0 + } +} + +/// One provider call of one run: the request's digest (never its text, so no +/// prompt or credential is stored), the reply exactly as the lane received it, +/// and what the call was charged. Keyed (run id, sequence). +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ProviderCallRecord { + run_id: String, + sequence: u64, + purpose: InferencePurpose, + model: String, + request_sha256: String, + events: Vec, + receipt_digest: String, + usage: Option, + charged: u64, +} + +impl ProviderCallRecord { + pub fn run_id(&self) -> &str { + &self.run_id + } + + pub fn sequence(&self) -> u64 { + self.sequence + } + + pub fn purpose(&self) -> InferencePurpose { + self.purpose + } + + pub fn model(&self) -> &str { + &self.model + } + + /// Lowercase hex of the provider request's digest. + pub fn request_sha256(&self) -> &str { + &self.request_sha256 + } + + pub fn usage(&self) -> Option { + self.usage + } + + /// Tokens this call was charged against the run's cap. + pub fn charged(&self) -> u64 { + self.charged + } + + /// The reply as the lane received it. + pub fn output(&self) -> InferenceOutput { + let output = InferenceOutput::new(self.events.clone(), self.receipt_digest.clone()); + match self.usage { + Some(usage) => output.with_usage(usage), + None => output, + } + } +} + +fn row_key(run: &str, sequence: u64) -> String { + format!("{run}/{sequence:020}") +} + +fn hex(digest: &[u8; 32]) -> String { + digest.iter().map(|byte| format!("{byte:02x}")).collect() +} + +fn decode_row(row: &CultCacheEnvelope) -> Result { + if row.r#type != CALL_RECORD_ROW || row.schema_id.as_deref() != Some(CALL_RECORD_SCHEMA) { + return Err(CallRecordError::Corrupt); + } + let record: ProviderCallRecord = + rmp_serde::from_slice(&row.payload).map_err(|_| CallRecordError::Corrupt)?; + let canonical = rmp_serde::to_vec_named(&record).map_err(|_| CallRecordError::Corrupt)?; + if canonical != row.payload + || RunId::new(&record.run_id).is_err() + || record.sequence == 0 + || row.key != row_key(&record.run_id, record.sequence) + { + return Err(CallRecordError::Corrupt); + } + Ok(record) +} + +#[derive(Default)] +struct RunTally { + last_sequence: u64, + count: u64, + charged: u64, +} + +struct StoreState { + store: OwnedRedbMessagePackBackingStore, + runs: BTreeMap, +} + +/// The call records of every run in one CultCache file. Records are appended +/// and never changed; a run's sequence is dense from 1. +pub struct CallRecordStore { + state: Mutex, +} + +impl CallRecordStore { + /// Opens the file, creating it empty when absent. Every row must be a + /// canonical call record of this schema with its key and a dense sequence; + /// anything else, an earlier or later schema version included, is refused + /// and nothing is migrated. + pub fn open(path: impl AsRef) -> Result { + let store = OwnedRedbMessagePackBackingStore::new(path.as_ref()) + .map_err(|_| CallRecordError::Store)?; + store + .validate_path_identity() + .map_err(|_| CallRecordError::Store)?; + let rows = store.pull_all().map_err(|_| CallRecordError::Store)?; + let mut runs: BTreeMap = BTreeMap::new(); + for row in &rows { + let record = decode_row(row)?; + let tally = runs.entry(record.run_id).or_default(); + tally.last_sequence = tally.last_sequence.max(record.sequence); + tally.count += 1; + tally.charged = tally.charged.saturating_add(record.charged); + } + if runs.values().any(|tally| tally.count != tally.last_sequence) { + return Err(CallRecordError::Corrupt); + } + Ok(Self { + state: Mutex::new(StoreState { store, runs }), + }) + } + + fn lock(&self) -> Result, CallRecordError> { + self.state.lock().map_err(|_| CallRecordError::Store) + } + + /// The sequence the next record of this run must carry. + pub fn next_sequence(&self, run: &RunId) -> Result { + Ok(self + .lock()? + .runs + .get(run.as_str()) + .map_or(0, |tally| tally.last_sequence) + + 1) + } + + /// Tokens this run's stored calls were charged. + pub fn charged(&self, run: &RunId) -> Result { + Ok(self + .lock()? + .runs + .get(run.as_str()) + .map_or(0, |tally| tally.charged)) + } + + fn append(&self, record: &ProviderCallRecord) -> Result<(), CallRecordError> { + let mut state = self.lock()?; + let last = state + .runs + .get(&record.run_id) + .map_or(0, |tally| tally.last_sequence); + if record.sequence != last + 1 { + return Err(CallRecordError::OutOfSequence); + } + let row = CultCacheEnvelope { + key: row_key(&record.run_id, record.sequence), + r#type: CALL_RECORD_ROW.into(), + payload: rmp_serde::to_vec_named(record).map_err(|_| CallRecordError::Store)?, + stored_at: Utc::now().to_rfc3339(), + schema_id: Some(CALL_RECORD_SCHEMA.into()), + }; + state.store.push(&row).map_err(|_| CallRecordError::Store)?; + let tally = state.runs.entry(record.run_id.clone()).or_default(); + tally.last_sequence = record.sequence; + tally.count += 1; + tally.charged = tally.charged.saturating_add(record.charged); + Ok(()) + } + + /// The run's records in sequence. + pub fn records(&self, run: &RunId) -> Result, CallRecordError> { + let prefix = format!("{}/", run.as_str()); + let rows = self + .lock()? + .store + .pull_all() + .map_err(|_| CallRecordError::Store)?; + let mut records = Vec::new(); + for row in rows.iter().filter(|row| row.key.starts_with(&prefix)) { + records.push(decode_row(row)?); + } + records.sort_by_key(|record| record.sequence); + Ok(records) + } +} + +struct Ledger { + /// Tokens charged by finished calls, restored from the store at open. + settled: u64, + /// Bounds held by calls still on the wire. + in_flight: u64, +} + +/// Decorates a port for one run: refuses a call that could pass the cap, +/// charges the real cost, and stores every reply before returning it. +pub struct RecordingInferencePort { + inner: Arc, + store: Arc, + run: RunId, + cap: u64, + ledger: Mutex, +} + +impl RecordingInferencePort { + /// The run's spend resumes from its stored records. + pub fn open( + inner: Arc, + store: Arc, + run: RunId, + cap_tokens: u64, + ) -> Result { + let settled = store.charged(&run)?; + Ok(Self { + inner, + store, + run, + cap: cap_tokens, + ledger: Mutex::new(Ledger { + settled, + in_flight: 0, + }), + }) + } + + /// Tokens charged to the run so far. + pub fn spent(&self) -> u64 { + self.ledger.lock().map_or(u64::MAX, |ledger| ledger.settled) + } + + /// The most this request can cost: its serialized bytes plus its output + /// allowance. A request with no output allowance has no bound, so it + /// cannot be held under a cap. + fn bound(request: &PreparedInference) -> Result { + let allowance = request + .invocation + .request + .max_output_tokens + .ok_or_else(InferenceFault::budget_exhausted)?; + let payload = serde_json::to_vec(&request.invocation.request) + .map_err(|_| InferenceFault::new("a provider request could not be measured"))?; + Ok((payload.len() as u64).saturating_add(u64::from(allowance))) + } + + fn reserve(&self, bound: u64) -> Result<(), InferenceFault> { + let mut ledger = self + .ledger + .lock() + .map_err(|_| InferenceFault::integrity_violation("the run's ledger was poisoned"))?; + let held = ledger.settled.saturating_add(ledger.in_flight); + if held.saturating_add(bound) > self.cap { + return Err(InferenceFault::budget_exhausted()); + } + ledger.in_flight += bound; + Ok(()) + } + + fn settle( + &self, + bound: u64, + identity: (InferencePurpose, String, String), + result: Result, + ) -> Result { + let mut ledger = self + .ledger + .lock() + .map_err(|_| InferenceFault::integrity_violation("the run's ledger was poisoned"))?; + ledger.in_flight = ledger.in_flight.saturating_sub(bound); + let output = match result { + Ok(output) => output, + Err(fault) => { + if fault.class() != InferenceFaultClass::Connect { + ledger.settled = ledger.settled.saturating_add(bound); + } + return Err(fault); + } + }; + let charged = match output.usage { + Some(usage) => usage.prompt.saturating_add(usage.completion), + None => bound, + }; + ledger.settled = ledger.settled.saturating_add(charged); + let (purpose, model, request_sha256) = identity; + let sequence = self.store.next_sequence(&self.run).map_err(unrecorded)?; + let record = ProviderCallRecord { + run_id: self.run.as_str().to_owned(), + sequence, + purpose, + model, + request_sha256, + events: output.events.clone(), + receipt_digest: output.receipt_digest.clone(), + usage: output.usage, + charged, + }; + self.store.append(&record).map_err(unrecorded)?; + Ok(output) + } +} + +fn unrecorded(_: CallRecordError) -> InferenceFault { + InferenceFault::integrity_violation("a provider call could not be recorded") +} + +#[async_trait] +impl InferencePort for RecordingInferencePort { + fn prepare(&self, request: InferenceRequest) -> Result { + self.inner.prepare(request) + } + + async fn infer(&self, request: PreparedInference) -> Result { + let bound = Self::bound(&request)?; + self.reserve(bound)?; + let identity = ( + request.purpose, + request.invocation.request.model.clone(), + hex(&request.invocation.provider_request_sha256), + ); + let result = self.inner.infer(request).await; + self.settle(bound, identity, result) + } + + fn lend_tool_results(&self, prepared: &PreparedInference, oracle: Box) { + self.inner.lend_tool_results(prepared, oracle); + } +} + +/// Serves a run's stored replies, in sequence, to the same requests. It holds +/// no provider and makes no call: a request that is not the next recorded one +/// is a fault. +pub struct ReplayInferencePort { + caller_runtime_id: String, + pending: Mutex>, +} + +impl ReplayInferencePort { + /// `records` are one run's, as `CallRecordStore::records` returns them. + pub fn new( + caller_runtime_id: impl Into, + records: Vec, + ) -> Result { + let dense = records.iter().enumerate().all(|(index, record)| { + record.sequence == index as u64 + 1 && record.run_id == records[0].run_id + }); + if !dense { + return Err(CallRecordError::NotReplayable); + } + Ok(Self { + caller_runtime_id: caller_runtime_id.into(), + pending: Mutex::new(records.into()), + }) + } +} + +#[async_trait] +impl InferencePort for ReplayInferencePort { + fn prepare(&self, request: InferenceRequest) -> Result { + PreparedInference::prepare( + &self.caller_runtime_id, + unix_ms()?.saturating_add(REQUEST_EXPIRY.as_millis() as u64), + request, + ) + } + + async fn infer(&self, request: PreparedInference) -> Result { + let mut pending = self + .pending + .lock() + .map_err(|_| InferenceFault::integrity_violation("the replay queue was poisoned"))?; + let digest = hex(&request.invocation.provider_request_sha256); + let Some(next) = pending.front() else { + return Err(InferenceFault::integrity_violation( + "replay has no recorded call left", + )); + }; + if next.request_sha256 != digest { + return Err(InferenceFault::integrity_violation( + "the request is not the next recorded call", + )); + } + let output = next.output(); + pending.pop_front(); + Ok(output) + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/ghostlight/src/verse/calls/tests.rs b/crates/ghostlight/src/verse/calls/tests.rs new file mode 100644 index 0000000..67b8b2a --- /dev/null +++ b/crates/ghostlight/src/verse/calls/tests.rs @@ -0,0 +1,652 @@ +use super::*; +use crate::CommandId; +use crate::controllers::{InferenceFaultDisposition, RequestShape, tool_request}; +use codex_connector::{CodexInputItem, CodexToolDefinition}; +use std::sync::atomic::{AtomicUsize, Ordering}; + +const MODEL: &str = "hosted/test-model"; + +type Script = Vec>; + +/// A provider stand-in: answers from a script, counts the calls that +/// reached it, and can hold a call open until released. +struct ScriptedPort { + replies: Mutex>>, + calls: AtomicUsize, + gate: Option>, +} + +impl ScriptedPort { + fn new(replies: Script) -> Arc { + Arc::new(Self { + replies: Mutex::new(replies.into()), + calls: AtomicUsize::new(0), + gate: None, + }) + } + + fn gated(replies: Script, gate: Arc) -> Arc { + Arc::new(Self { + replies: Mutex::new(replies.into()), + calls: AtomicUsize::new(0), + gate: Some(gate), + }) + } + + fn calls(&self) -> usize { + self.calls.load(Ordering::SeqCst) + } +} + +#[async_trait] +impl InferencePort for ScriptedPort { + fn prepare(&self, request: InferenceRequest) -> Result { + PreparedInference::prepare("verse-test", 4_102_444_800_000, request) + } + + async fn infer(&self, _: PreparedInference) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + if let Some(gate) = &self.gate { + gate.acquire().await.expect("the gate stays open").forget(); + } + self.replies + .lock() + .unwrap() + .pop_front() + .expect("the script has a reply for every call that reaches it") + } +} + +fn request_for(command: CommandId, prompt: &str, max_output_tokens: u32) -> InferenceRequest { + tool_request( + command, + 0, + InferencePurpose::Persona, + MODEL, + "Respond only in prose.", + vec![CodexInputItem::UserText { + text: prompt.into(), + }], + Vec::::new(), + RequestShape { + max_output_tokens, + parallel_tool_calls: false, + }, + ) + .expect("the request builds") +} + +fn prepared_for(prompt: &str, max_output_tokens: u32) -> PreparedInference { + PreparedInference::prepare( + "verse-test", + 4_102_444_800_000, + request_for(CommandId::new(), prompt, max_output_tokens), + ) + .expect("the request prepares") +} + +/// What the spec defines the bound as, computed from the prepared request. +fn bound_of(prepared: &PreparedInference) -> u64 { + let payload = serde_json::to_vec(&prepared.invocation.request) + .expect("a provider request serializes") + .len() as u64; + payload + + u64::from( + prepared + .invocation + .request + .max_output_tokens + .expect("an allowance"), + ) +} + +fn reply(text: &str, receipt: &str) -> InferenceOutput { + InferenceOutput::new(vec![InferenceEvent::Text(text.into())], receipt) +} + +fn usage(prompt: u64, completion: u64, cached_prompt: Option) -> TokenUsage { + TokenUsage { + prompt, + completion, + cached_prompt, + } +} + +fn store_in(directory: &tempfile::TempDir) -> (Arc, std::path::PathBuf) { + let path = directory.path().join("calls.redb"); + ( + Arc::new(CallRecordStore::open(&path).expect("the store opens")), + path, + ) +} + +fn recording( + inner: &Arc, + store: &Arc, + run: &str, + cap: u64, +) -> RecordingInferencePort { + RecordingInferencePort::open(inner.clone(), store.clone(), RunId::new(run).unwrap(), cap) + .expect("the port opens") +} + +fn json(output: &InferenceOutput) -> Vec { + serde_json::to_vec(output).expect("an output serializes") +} + +fn ok_all(outputs: &[InferenceOutput]) -> Script { + outputs.iter().cloned().map(Ok).collect() +} + +#[tokio::test] +async fn recording_port_stores_each_call_before_returning() { + let directory = tempfile::tempdir().unwrap(); + let (store, path) = store_in(&directory); + let run = RunId::new("run-a").unwrap(); + let outputs = vec![ + reply(" first\nreply \u{e9}\u{1f9ff} ", "receipt-1").with_usage(usage(100, 20, Some(40))), + InferenceOutput::new( + vec![ + InferenceEvent::Text("a".into()), + InferenceEvent::ToolCall { + call_id: "c1".into(), + name: "speak".into(), + arguments: "{\"b\":1, \"a\":[1,2]}".into(), + }, + ], + "receipt-2", + ), + InferenceOutput::new(Vec::new(), "receipt-3").with_usage(usage(7, 0, None)), + ]; + let inner = ScriptedPort::new(ok_all(&outputs)); + let port = recording(&inner, &store, "run-a", u64::MAX); + for (index, expected) in outputs.iter().enumerate() { + let prepared = port + .prepare(request_for(CommandId::new(), "tell", 1_200)) + .unwrap(); + let digest = hex(&prepared.invocation.provider_request_sha256); + let returned = port.infer(prepared).await.unwrap(); + assert_eq!(json(&returned), json(expected)); + let stored = store.records(&run).unwrap(); + assert_eq!( + stored.len(), + index + 1, + "the call was stored before it returned" + ); + let record = &stored[index]; + assert_eq!(record.sequence(), index as u64 + 1); + assert_eq!(record.run_id(), "run-a"); + assert_eq!(record.request_sha256(), digest); + assert_eq!(record.model(), MODEL); + assert_eq!(record.purpose(), InferencePurpose::Persona); + assert_eq!(record.usage(), expected.usage); + assert_eq!(json(&record.output()), json(expected), "byte for byte"); + } + // Prompt plus completion; the cached figure is part of the prompt. + let charges: Vec = store + .records(&run) + .unwrap() + .iter() + .map(|record| record.charged()) + .collect(); + assert_eq!(charges[0], 120); + assert_eq!(charges[2], 7); + assert_eq!(port.spent(), charges.iter().sum::()); + + // Another run keys its own sequence from 1. + let other = recording(&inner, &store, "run-b", u64::MAX); + inner + .replies + .lock() + .unwrap() + .push_back(Ok(reply("b", "receipt-b"))); + other + .infer( + other + .prepare(request_for(CommandId::new(), "tell", 1_200)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + store.records(&RunId::new("run-b").unwrap()).unwrap()[0].sequence(), + 1 + ); + + // The records are in the file, not in the port. + let before = store.records(&run).unwrap(); + drop((port, other, store)); + let reopened = CallRecordStore::open(&path).unwrap(); + assert_eq!(reopened.records(&run).unwrap(), before); +} + +#[tokio::test] +async fn cap_refuses_before_the_provider_is_called() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let probe = prepared_for("probe", 1_200); + let bound = bound_of(&probe); + assert!(bound > 1_200, "the bound includes the request's bytes"); + + let inner = ScriptedPort::new(vec![Ok(reply("never", "r"))]); + let port = recording(&inner, &store, "run-cap", bound - 1); + let fault = port.infer(probe.clone()).await.expect_err("over the cap"); + assert_eq!(fault.class(), InferenceFaultClass::BudgetExhausted); + assert_eq!( + fault.disposition(), + InferenceFaultDisposition::IntegrityViolation + ); + assert_eq!(inner.calls(), 0, "the provider was never invoked"); + assert!( + store + .records(&RunId::new("run-cap").unwrap()) + .unwrap() + .is_empty() + ); + assert_eq!(port.spent(), 0); + + // Exactly the cap is allowed. + let exact = recording(&inner, &store, "run-cap", bound); + exact + .infer(probe) + .await + .expect("a call that fits the cap is made"); + assert_eq!(inner.calls(), 1); + + // A request with no output allowance has no bound and is refused. + let mut open_ended = prepared_for("open ended", 1_200); + open_ended.invocation.request.max_output_tokens = None; + let unbounded = recording(&inner, &store, "run-open", u64::MAX); + let fault = unbounded + .infer(open_ended) + .await + .expect_err("no allowance, no bound"); + assert_eq!(fault.class(), InferenceFaultClass::BudgetExhausted); + assert_eq!(inner.calls(), 1); +} + +#[tokio::test] +async fn spend_never_exceeds_cap() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let run = RunId::new("run-spend").unwrap(); + let prompt = "a moderately long prompt for the sample call"; + let bound = bound_of(&prepared_for(prompt, 600)); + let cap = bound * 5 / 2; + let script: Script = (0..40) + .map(|n| Ok(reply("ok", &format!("r{n}")).with_usage(usage(bound / 4, bound / 8, None)))) + .collect(); + let inner = ScriptedPort::new(script); + let port = recording(&inner, &store, "run-spend", cap); + let mut made = 0; + loop { + match port.infer(prepared_for(prompt, 600)).await { + Ok(_) => made += 1, + Err(fault) => { + assert_eq!(fault.class(), InferenceFaultClass::BudgetExhausted); + break; + } + } + assert!(made < 40, "the cap never refused"); + } + assert!(made >= 2, "the cap let some calls through: {made}"); + assert_eq!( + inner.calls(), + made, + "a refused call never reached the provider" + ); + let total: u64 = store + .records(&run) + .unwrap() + .iter() + .map(|record| record.charged()) + .sum(); + assert!(total <= cap, "{total} > {cap}"); + assert_eq!(total, port.spent()); + // The refusal is the cap's doing: another call would not have fit. + assert!(total + bound > cap); +} + +#[tokio::test] +async fn calls_on_the_wire_hold_their_bound() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let bound = bound_of(&prepared_for("held", 1_200)); + let gate = Arc::new(tokio::sync::Semaphore::new(0)); + let inner = ScriptedPort::gated(vec![Ok(reply("one", "r1"))], gate.clone()); + // Room for one call's bound and not two. + let port = Arc::new(recording(&inner, &store, "run-flight", bound + bound / 2)); + let first = { + let port = port.clone(); + tokio::spawn(async move { port.infer(prepared_for("held", 1_200)).await }) + }; + while inner.calls() == 0 { + tokio::task::yield_now().await; + } + let fault = port + .infer(prepared_for("held", 1_200)) + .await + .expect_err("the first call's bound is still held"); + assert_eq!(fault.class(), InferenceFaultClass::BudgetExhausted); + assert_eq!(inner.calls(), 1); + gate.add_permits(1); + first.await.unwrap().expect("the first call completes"); +} + +#[tokio::test] +async fn a_reply_with_no_usage_is_charged_its_bound() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let prepared = prepared_for("no usage reported", 1_200); + let bound = bound_of(&prepared); + let inner = ScriptedPort::new(vec![Ok(reply("silent", "r"))]); + let port = recording(&inner, &store, "run-silent", u64::MAX); + port.infer(prepared).await.unwrap(); + let record = store + .records(&RunId::new("run-silent").unwrap()) + .unwrap() + .remove(0); + assert_eq!(record.usage(), None); + assert_eq!(record.charged(), bound); + assert_eq!(port.spent(), bound); +} + +#[tokio::test] +async fn a_fault_is_charged_its_bound_unless_it_never_connected() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let prepared = prepared_for("fault", 1_200); + let bound = bound_of(&prepared); + let inner = ScriptedPort::new(vec![ + Err(InferenceFault::retryable("down").classed(InferenceFaultClass::Connect)), + Err(InferenceFault::retryable("slow").classed(InferenceFaultClass::Timeout)), + ]); + let port = recording(&inner, &store, "run-fault", u64::MAX); + port.infer(prepared.clone()) + .await + .expect_err("connect fault"); + assert_eq!(port.spent(), 0, "nothing was sent"); + let fault = port.infer(prepared).await.expect_err("timeout fault"); + assert_eq!( + fault.class(), + InferenceFaultClass::Timeout, + "the fault passes through" + ); + assert_eq!( + port.spent(), + bound, + "a request that may have run is charged its bound" + ); + assert!( + store + .records(&RunId::new("run-fault").unwrap()) + .unwrap() + .is_empty() + ); +} + +#[tokio::test] +async fn a_reopened_run_resumes_its_spend() { + let directory = tempfile::tempdir().unwrap(); + let (store, path) = store_in(&directory); + let probe = prepared_for("resume", 1_200); + let bound = bound_of(&probe); + let inner = ScriptedPort::new(vec![Ok( + reply("one", "r1").with_usage(usage(bound, 0, None)) + )]); + let first = recording(&inner, &store, "run-resume", bound * 2); + first.infer(probe.clone()).await.unwrap(); + drop((first, store)); + + let store = Arc::new(CallRecordStore::open(&path).unwrap()); + let inner = ScriptedPort::new(vec![Ok(reply("two", "r2"))]); + let resumed = recording(&inner, &store, "run-resume", bound * 2 - 1); + assert_eq!(resumed.spent(), bound); + let fault = resumed + .infer(probe) + .await + .expect_err("the earlier spend counts"); + assert_eq!(fault.class(), InferenceFaultClass::BudgetExhausted); + assert_eq!(inner.calls(), 0); +} + +#[tokio::test] +async fn replay_serves_stored_responses_and_refuses_strangers() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let run = RunId::new("run-replay").unwrap(); + let commands = [CommandId::new(), CommandId::new(), CommandId::new()]; + let outputs = [ + reply("one \u{e9}", "r1").with_usage(usage(5, 6, Some(1))), + InferenceOutput::new( + vec![InferenceEvent::ToolCall { + call_id: "c".into(), + name: "speak".into(), + arguments: "{ \"x\" : 1 }".into(), + }], + "r2", + ), + reply("three", "r3"), + ]; + let inner = ScriptedPort::new(ok_all(&outputs)); + let live = recording(&inner, &store, "run-replay", u64::MAX); + for (command, prompt) in commands.iter().zip(["a", "b", "c"]) { + let prepared = live.prepare(request_for(*command, prompt, 1_200)).unwrap(); + live.infer(prepared).await.unwrap(); + } + let calls_made = inner.calls(); + + let stored = store.records(&run).unwrap(); + let replay = ReplayInferencePort::new("verse-replay", stored).unwrap(); + // A request the run never made is a fault and consumes nothing. + let stranger = replay + .prepare(request_for(CommandId::new(), "a", 1_200)) + .unwrap(); + let fault = replay.infer(stranger).await.expect_err("a stranger"); + assert_eq!( + fault.disposition(), + InferenceFaultDisposition::IntegrityViolation + ); + // Out of order is a stranger too. + let second_first = replay + .prepare(request_for(commands[1], "b", 1_200)) + .unwrap(); + replay + .infer(second_first) + .await + .expect_err("out of order"); + // In order, each request gets exactly what the run received. + for ((command, prompt), expected) in commands.iter().zip(["a", "b", "c"]).zip(&outputs) { + let prepared = replay + .prepare(request_for(*command, prompt, 1_200)) + .unwrap(); + let served = replay.infer(prepared).await.unwrap(); + assert_eq!(json(&served), json(expected)); + } + let again = replay + .prepare(request_for(commands[0], "a", 1_200)) + .unwrap(); + replay.infer(again).await.expect_err("the record is spent"); + assert_eq!(inner.calls(), calls_made, "replay made no provider call"); +} + +fn sample_record(run: &str, sequence: u64) -> ProviderCallRecord { + ProviderCallRecord { + run_id: run.into(), + sequence, + purpose: InferencePurpose::Persona, + model: MODEL.into(), + request_sha256: "ab".into(), + events: vec![InferenceEvent::Text("t".into())], + receipt_digest: "r".into(), + usage: None, + charged: 3, + } +} + +#[test] +fn replay_needs_one_runs_dense_records() { + assert!( + ReplayInferencePort::new("x", vec![sample_record("a", 1), sample_record("a", 2)]).is_ok() + ); + for records in [ + vec![sample_record("a", 2)], + vec![sample_record("a", 1), sample_record("a", 3)], + vec![sample_record("a", 1), sample_record("b", 2)], + ] { + assert_eq!( + ReplayInferencePort::new("x", records).err(), + Some(CallRecordError::NotReplayable) + ); + } +} + +#[tokio::test] +async fn a_record_holds_no_request_text() { + let directory = tempfile::tempdir().unwrap(); + let (store, path) = store_in(&directory); + let canary = "REQUEST-CANARY-91c4"; + let inner = ScriptedPort::new(vec![Ok(reply("fine", "r"))]); + let port = recording(&inner, &store, "run-secret", u64::MAX); + port.infer(prepared_for(canary, 1_200)).await.unwrap(); + drop((port, store)); + let raw = OwnedRedbMessagePackBackingStore::new(&path) + .unwrap() + .pull_all() + .unwrap(); + assert_eq!(raw.len(), 1); + for row in raw { + let bytes = [row.key.as_bytes(), row.payload.as_slice()].concat(); + assert!( + !bytes + .windows(canary.len()) + .any(|window| window == canary.as_bytes()), + "the request text reached the store" + ); + } + let refusal = InferenceFault::budget_exhausted(); + assert!(!format!("{refusal} {refusal:?}").contains(canary)); +} + +fn envelope( + row_type: &str, + schema: Option<&str>, + key: &str, + payload: Vec, +) -> CultCacheEnvelope { + CultCacheEnvelope { + key: key.into(), + r#type: row_type.into(), + payload, + stored_at: Utc::now().to_rfc3339(), + schema_id: schema.map(str::to_owned), + } +} + +#[test] +fn the_store_refuses_any_row_that_is_not_a_canonical_v1_record() { + let good = sample_record("run-x", 1); + let payload = rmp_serde::to_vec_named(&good).unwrap(); + let key = row_key("run-x", 1); + let mut loose = payload.clone(); + loose.push(0xc0); + let v1 = |key: &str, payload: Vec| { + envelope(CALL_RECORD_ROW, Some(CALL_RECORD_SCHEMA), key, payload) + }; + let record_at = |sequence: u64| { + v1( + &row_key("run-x", sequence), + rmp_serde::to_vec_named(&sample_record("run-x", sequence)).unwrap(), + ) + }; + let cases = [ + ( + "earlier version", + envelope( + "verse_provider_call.v0", + Some("ghostlight.verse_provider_call.v0"), + &key, + payload.clone(), + ), + ), + ( + "later version", + envelope( + "verse_provider_call.v2", + Some("ghostlight.verse_provider_call.v2"), + &key, + payload.clone(), + ), + ), + ( + "another document", + envelope( + "controller_work.v18", + Some("ghostlight.controller_work.v18"), + &key, + payload.clone(), + ), + ), + ( + "no schema", + envelope(CALL_RECORD_ROW, None, &key, payload.clone()), + ), + ("wrong key", v1("run-x/2", payload.clone())), + ("not canonical", v1(&key, loose)), + ("not a record", v1(&key, vec![1, 2, 3])), + ("sequence zero", record_at(0)), + ("a hole in the run", record_at(2)), + ]; + for (label, row) in cases { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("calls.redb"); + OwnedRedbMessagePackBackingStore::new(&path) + .unwrap() + .push(&row) + .unwrap(); + let error = CallRecordStore::open(&path).err(); + assert_eq!(error, Some(CallRecordError::Corrupt), "{label}"); + } + // The same row, well formed, opens. + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("calls.redb"); + OwnedRedbMessagePackBackingStore::new(&path) + .unwrap() + .push(&v1(&key, payload)) + .unwrap(); + let store = CallRecordStore::open(&path).unwrap(); + let run = RunId::new("run-x").unwrap(); + assert_eq!(store.records(&run).unwrap(), vec![good]); + assert_eq!(store.charged(&run).unwrap(), 3); +} + +#[test] +fn a_record_is_appended_only_at_the_next_sequence() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let run = RunId::new("run-seq").unwrap(); + assert_eq!( + store.append(&sample_record("run-seq", 2)), + Err(CallRecordError::OutOfSequence) + ); + store.append(&sample_record("run-seq", 1)).unwrap(); + assert_eq!( + store.append(&sample_record("run-seq", 1)), + Err(CallRecordError::OutOfSequence) + ); + assert_eq!( + store.append(&sample_record("run-seq", 3)), + Err(CallRecordError::OutOfSequence) + ); + store.append(&sample_record("run-seq", 2)).unwrap(); + assert_eq!(store.next_sequence(&run), Ok(3)); +} + +#[test] +fn a_run_id_is_a_key_safe_segment() { + assert!(RunId::new("run_1-A").is_ok()); + assert!(RunId::new(&"a".repeat(64)).is_ok()); + for bad in ["", "has/slash", "has space", "caf\u{e9}", &"a".repeat(65)] { + assert_eq!(RunId::new(bad), Err(CallRecordError::InvalidRunId)); + } +} diff --git a/crates/ghostlight/src/verse/mod.rs b/crates/ghostlight/src/verse/mod.rs new file mode 100644 index 0000000..5c1e308 --- /dev/null +++ b/crates/ghostlight/src/verse/mod.rs @@ -0,0 +1,10 @@ +//! The Aetheria verse run's own organs. The verse binary and its run record +//! live above this module; here sit the parts of a run that belong to the +//! library: the record of every provider call and the spend it implies. + +mod calls; + +pub use calls::{ + CALL_RECORD_SCHEMA, CallRecordError, CallRecordStore, ProviderCallRecord, RecordingInferencePort, + ReplayInferencePort, RunId, +}; From f74dec6675d13e1bd8112e1d5508216ed81e32b8 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 10 Oct 2026 14:54:33 +0200 Subject: [PATCH 03/10] Verse call records: kill the first mutation round's survivors The in-memory run tally no longer counts rows (only open used it); the store test pins a record whose run id is not a key segment; the lend_tool_results forwarding has a test; the in-flight test fails by timeout instead of hanging when a held call is let through. Builds: yes. Co-Authored-By: Claude Opus 5.5 --- crates/ghostlight/src/verse/calls.rs | 9 ++-- crates/ghostlight/src/verse/calls/tests.rs | 48 ++++++++++++++++++++-- 2 files changed, 49 insertions(+), 8 deletions(-) diff --git a/crates/ghostlight/src/verse/calls.rs b/crates/ghostlight/src/verse/calls.rs index 328417b..21d5c72 100644 --- a/crates/ghostlight/src/verse/calls.rs +++ b/crates/ghostlight/src/verse/calls.rs @@ -164,7 +164,6 @@ fn decode_row(row: &CultCacheEnvelope) -> Result = BTreeMap::new(); + let mut counts: BTreeMap = BTreeMap::new(); for row in &rows { let record = decode_row(row)?; + *counts.entry(record.run_id.clone()).or_default() += 1; let tally = runs.entry(record.run_id).or_default(); tally.last_sequence = tally.last_sequence.max(record.sequence); - tally.count += 1; tally.charged = tally.charged.saturating_add(record.charged); } - if runs.values().any(|tally| tally.count != tally.last_sequence) { + // Keys are unique, so a run is dense from 1 exactly when its rows + // number its highest sequence. + if runs.iter().any(|(run, tally)| counts[run] != tally.last_sequence) { return Err(CallRecordError::Corrupt); } Ok(Self { @@ -249,7 +251,6 @@ impl CallRecordStore { state.store.push(&row).map_err(|_| CallRecordError::Store)?; let tally = state.runs.entry(record.run_id.clone()).or_default(); tally.last_sequence = record.sequence; - tally.count += 1; tally.charged = tally.charged.saturating_add(record.charged); Ok(()) } diff --git a/crates/ghostlight/src/verse/calls/tests.rs b/crates/ghostlight/src/verse/calls/tests.rs index 67b8b2a..357fb59 100644 --- a/crates/ghostlight/src/verse/calls/tests.rs +++ b/crates/ghostlight/src/verse/calls/tests.rs @@ -13,6 +13,7 @@ type Script = Vec>; struct ScriptedPort { replies: Mutex>>, calls: AtomicUsize, + lent: AtomicUsize, gate: Option>, } @@ -21,6 +22,7 @@ impl ScriptedPort { Arc::new(Self { replies: Mutex::new(replies.into()), calls: AtomicUsize::new(0), + lent: AtomicUsize::new(0), gate: None, }) } @@ -29,6 +31,7 @@ impl ScriptedPort { Arc::new(Self { replies: Mutex::new(replies.into()), calls: AtomicUsize::new(0), + lent: AtomicUsize::new(0), gate: Some(gate), }) } @@ -44,6 +47,10 @@ impl InferencePort for ScriptedPort { PreparedInference::prepare("verse-test", 4_102_444_800_000, request) } + fn lend_tool_results(&self, _: &PreparedInference, _: Box) { + self.lent.fetch_add(1, Ordering::SeqCst); + } + async fn infer(&self, _: PreparedInference) -> Result { self.calls.fetch_add(1, Ordering::SeqCst); if let Some(gate) = &self.gate { @@ -323,16 +330,42 @@ async fn calls_on_the_wire_hold_their_bound() { while inner.calls() == 0 { tokio::task::yield_now().await; } - let fault = port - .infer(prepared_for("held", 1_200)) - .await - .expect_err("the first call's bound is still held"); + // A second call that was let through would wait on the closed gate. + let second = tokio::time::timeout( + std::time::Duration::from_secs(5), + port.infer(prepared_for("held", 1_200)), + ) + .await + .expect("the second call was refused, not left waiting on the provider"); + let fault = second.expect_err("the first call's bound is still held"); assert_eq!(fault.class(), InferenceFaultClass::BudgetExhausted); assert_eq!(inner.calls(), 1); gate.add_permits(1); first.await.unwrap().expect("the first call completes"); } +struct NoTools; + +impl ToolResultOracle for NoTools { + fn remaining_rounds(&self) -> u32 { + 0 + } + + fn answer(&mut self, _: &str, _: &str) -> Result { + Ok(String::new()) + } +} + +#[test] +fn tool_results_are_lent_to_the_port_underneath() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let inner = ScriptedPort::new(Vec::new()); + let port = recording(&inner, &store, "run-lend", u64::MAX); + port.lend_tool_results(&prepared_for("lend", 1_200), Box::new(NoTools)); + assert_eq!(inner.lent.load(Ordering::SeqCst), 1); +} + #[tokio::test] async fn a_reply_with_no_usage_is_charged_its_bound() { let directory = tempfile::tempdir().unwrap(); @@ -594,6 +627,13 @@ fn the_store_refuses_any_row_that_is_not_a_canonical_v1_record() { ("wrong key", v1("run-x/2", payload.clone())), ("not canonical", v1(&key, loose)), ("not a record", v1(&key, vec![1, 2, 3])), + ( + "a run id that is not a key segment", + v1( + &row_key("bad/run", 1), + rmp_serde::to_vec_named(&sample_record("bad/run", 1)).unwrap(), + ), + ), ("sequence zero", record_at(0)), ("a hole in the run", record_at(2)), ]; From 2e9905f42f0f214e2d7c40c5346f80f114a2305d Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 10 Oct 2026 15:39:15 +0200 Subject: [PATCH 04/10] Verse call records: the in-flight test cannot hang when the first call is refused Two reserve mutants timed out because the test spun waiting for a call a mutant had refused. The wait is now bounded and fails. Builds: yes. Co-Authored-By: Claude Opus 5.5 --- crates/ghostlight/src/verse/calls/tests.rs | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/crates/ghostlight/src/verse/calls/tests.rs b/crates/ghostlight/src/verse/calls/tests.rs index 357fb59..eaecd1b 100644 --- a/crates/ghostlight/src/verse/calls/tests.rs +++ b/crates/ghostlight/src/verse/calls/tests.rs @@ -327,9 +327,13 @@ async fn calls_on_the_wire_hold_their_bound() { let port = port.clone(); tokio::spawn(async move { port.infer(prepared_for("held", 1_200)).await }) }; - while inner.calls() == 0 { - tokio::task::yield_now().await; - } + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while inner.calls() == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("the first call reached the provider"); // A second call that was let through would wait on the closed gate. let second = tokio::time::timeout( std::time::Duration::from_secs(5), From bd8dc756d91afd15c7ecb87d3c33b5ed8596e23a Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 10 Oct 2026 16:45:18 +0200 Subject: [PATCH 05/10] WIP call-record fix1: run ledger owned by the store (does not build until verified) Co-Authored-By: Claude Opus 5.5 --- crates/ghostlight/src/verse/calls.rs | 364 +++++++++++++++------ crates/ghostlight/src/verse/calls/tests.rs | 296 +++++++++++++++-- crates/ghostlight/src/verse/mod.rs | 4 +- 3 files changed, 522 insertions(+), 142 deletions(-) diff --git a/crates/ghostlight/src/verse/calls.rs b/crates/ghostlight/src/verse/calls.rs index 21d5c72..cfd3460 100644 --- a/crates/ghostlight/src/verse/calls.rs +++ b/crates/ghostlight/src/verse/calls.rs @@ -10,33 +10,37 @@ //! Lanes and runners never count tokens. A request is held against the cap by //! its bound: the serialized provider request's bytes (a token is at least one //! byte, so bytes bound the prompt) plus the request's own output allowance. -//! The bound is exact for a port that makes one provider round per call (the -//! hosted and local lanes). A port that loops tools inside one call can spend -//! more than one round's bound; its settled usage is still charged, so the next -//! call is refused, but this decorator cannot stop the overshoot of that call. //! -//! Charging rule, one path: a reply that reports usage is charged prompt plus -//! completion (cached prompt tokens are part of the prompt); a reply that -//! reports none is charged its bound; a fault is charged its bound, except a -//! connect fault, which never reached the provider and costs nothing. +//! The run's spend is one persisted ledger row per run, owned by the store and +//! written where it is spent. A call reserves its bound in that row before the +//! provider is called; a reply settles the reservation to what it reported +//! (prompt plus completion, cached prompt tokens being part of the prompt; none +//! reported is charged the bound); a fault leaves the reservation charged. A +//! restart reads the row, so no restart returns budget. A reply that reports +//! more than the call's bound is charged in full and faults the call, so the +//! run's spend never passes the cap unnoticed. One port at a time owns a run. use crate::controllers::{ - InferenceEvent, InferenceFault, InferenceFaultClass, InferenceOutput, InferencePort, - InferencePurpose, InferenceRequest, PreparedInference, REQUEST_EXPIRY, TokenUsage, - ToolResultOracle, unix_ms, + InferenceEvent, InferenceFault, InferenceOutput, InferencePort, InferencePurpose, + InferenceRequest, PreparedInference, REQUEST_EXPIRY, TokenUsage, ToolResultOracle, unix_ms, }; use async_trait::async_trait; use chrono::Utc; +use codex_connector::CodexProviderRequest; use cultcache_rs::{CacheBackingStore, CultCacheEnvelope, OwnedRedbMessagePackBackingStore}; use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; use std::{ - collections::{BTreeMap, VecDeque}, + collections::{BTreeMap, BTreeSet, VecDeque}, path::Path, sync::{Arc, Mutex}, }; use thiserror::Error; const CALL_RECORD_ROW: &str = "verse_provider_call.v1"; +const RUN_SPEND_ROW: &str = "verse_run_spend.v1"; +/// The run ledger document's schema. +const RUN_SPEND_SCHEMA: &str = "ghostlight.verse_run_spend.v1"; /// The call record document's schema. A store holding any other row type or /// schema is refused at open: this store is not migrated from, or shared with, /// another document. @@ -54,6 +58,10 @@ pub enum CallRecordError { OutOfSequence, #[error("replay needs one run's records in dense sequence from 1")] NotReplayable, + #[error("another recording port already owns this run")] + RunInUse, + #[error("the call could pass the run's token cap")] + OverCap, } /// A run's identity as a key segment. @@ -79,7 +87,7 @@ impl RunId { } } -/// One provider call of one run: the request's digest (never its text, so no +/// One provider call of one run: the request's content digest (never its text, so no /// prompt or credential is stored), the reply exactly as the lane received it, /// and what the call was charged. Keyed (run id, sequence). #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -88,7 +96,7 @@ pub struct ProviderCallRecord { sequence: u64, purpose: InferencePurpose, model: String, - request_sha256: String, + content_sha256: String, events: Vec, receipt_digest: String, usage: Option, @@ -112,9 +120,9 @@ impl ProviderCallRecord { &self.model } - /// Lowercase hex of the provider request's digest. - pub fn request_sha256(&self) -> &str { - &self.request_sha256 + /// Lowercase hex of the request's content digest (see `content_digest`). + pub fn content_sha256(&self) -> &str { + &self.content_sha256 } pub fn usage(&self) -> Option { @@ -140,49 +148,110 @@ fn row_key(run: &str, sequence: u64) -> String { format!("{run}/{sequence:020}") } -fn hex(digest: &[u8; 32]) -> String { +fn hex(digest: &[u8]) -> String { digest.iter().map(|byte| format!("{byte:02x}")).collect() } -fn decode_row(row: &CultCacheEnvelope) -> Result { - if row.r#type != CALL_RECORD_ROW || row.schema_id.as_deref() != Some(CALL_RECORD_SCHEMA) { - return Err(CallRecordError::Corrupt); +/// What a request asks of the provider, without the ids that name it. The +/// request and conversation ids are derived from the command id, and a command +/// id from the world it ran in, so keying on them would tie a recorded run to +/// one world's random identity. Everything else the provider is sent +/// (model, instructions, input, tools, shape) is in the digest. +fn content_digest(request: &CodexProviderRequest) -> Result { + let bytes = rmp_serde::to_vec(&( + &request.schema_id, + &request.model, + &request.instructions, + &request.input, + &request.reasoning_effort, + &request.reasoning_summary, + &request.service_tier, + &request.output_format_name, + &request.previous_response_id, + &request.tools, + &request.tool_choice, + request.parallel_tool_calls, + &request.output_schema_json, + request.max_output_tokens, + &request.prompt_cache_key, + )) + .map_err(|_| InferenceFault::new("a provider request could not be measured"))?; + Ok(hex(&Sha256::digest(bytes))) +} + +/// A run's persisted ledger: tokens reserved or charged against its cap. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct RunSpend { + run_id: String, + spent: u64, +} + +enum DecodedRow { + Call(ProviderCallRecord), + Spend(RunSpend), +} + +fn canonical Deserialize<'de>>( + row: &CultCacheEnvelope, +) -> Result { + let value: T = rmp_serde::from_slice(&row.payload).map_err(|_| CallRecordError::Corrupt)?; + let again = rmp_serde::to_vec_named(&value).map_err(|_| CallRecordError::Corrupt)?; + if again == row.payload { + Ok(value) + } else { + Err(CallRecordError::Corrupt) } - let record: ProviderCallRecord = - rmp_serde::from_slice(&row.payload).map_err(|_| CallRecordError::Corrupt)?; - let canonical = rmp_serde::to_vec_named(&record).map_err(|_| CallRecordError::Corrupt)?; - if canonical != row.payload - || RunId::new(&record.run_id).is_err() - || record.sequence == 0 - || row.key != row_key(&record.run_id, record.sequence) - { - return Err(CallRecordError::Corrupt); +} + +fn decode_row(row: &CultCacheEnvelope) -> Result { + match (row.r#type.as_str(), row.schema_id.as_deref()) { + (CALL_RECORD_ROW, Some(CALL_RECORD_SCHEMA)) => { + let record: ProviderCallRecord = canonical(row)?; + if RunId::new(&record.run_id).is_err() + || record.sequence == 0 + || row.key != row_key(&record.run_id, record.sequence) + { + return Err(CallRecordError::Corrupt); + } + Ok(DecodedRow::Call(record)) + } + (RUN_SPEND_ROW, Some(RUN_SPEND_SCHEMA)) => { + let spend: RunSpend = canonical(row)?; + if RunId::new(&spend.run_id).is_err() || row.key != spend.run_id { + return Err(CallRecordError::Corrupt); + } + Ok(DecodedRow::Spend(spend)) + } + _ => Err(CallRecordError::Corrupt), } - Ok(record) } #[derive(Default)] struct RunTally { last_sequence: u64, - charged: u64, + /// Tokens reserved or charged: the persisted ledger, read at open. + spent: u64, } struct StoreState { store: OwnedRedbMessagePackBackingStore, runs: BTreeMap, + /// Runs with a recording port alive; one ledger has one owner. + owned: BTreeSet, } -/// The call records of every run in one CultCache file. Records are appended -/// and never changed; a run's sequence is dense from 1. +/// The call records and spend ledgers of every run in one CultCache file. +/// Records are appended and never changed; a run's sequence is dense from 1. pub struct CallRecordStore { state: Mutex, } impl CallRecordStore { /// Opens the file, creating it empty when absent. Every row must be a - /// canonical call record of this schema with its key and a dense sequence; - /// anything else, an earlier or later schema version included, is refused - /// and nothing is migrated. + /// canonical call record or run ledger of this schema with its key, a + /// dense sequence, and a ledger that covers its records' charges; anything + /// else, an earlier or later schema version included, is refused and + /// nothing is migrated. pub fn open(path: impl AsRef) -> Result { let store = OwnedRedbMessagePackBackingStore::new(path.as_ref()) .map_err(|_| CallRecordError::Store)?; @@ -192,20 +261,37 @@ impl CallRecordStore { let rows = store.pull_all().map_err(|_| CallRecordError::Store)?; let mut runs: BTreeMap = BTreeMap::new(); let mut counts: BTreeMap = BTreeMap::new(); + let mut charged: BTreeMap = BTreeMap::new(); for row in &rows { - let record = decode_row(row)?; - *counts.entry(record.run_id.clone()).or_default() += 1; - let tally = runs.entry(record.run_id).or_default(); - tally.last_sequence = tally.last_sequence.max(record.sequence); - tally.charged = tally.charged.saturating_add(record.charged); + match decode_row(row)? { + DecodedRow::Call(record) => { + *counts.entry(record.run_id.clone()).or_default() += 1; + let total = charged.entry(record.run_id.clone()).or_default(); + *total = total.saturating_add(record.charged); + let tally = runs.entry(record.run_id).or_default(); + tally.last_sequence = tally.last_sequence.max(record.sequence); + } + DecodedRow::Spend(spend) => { + runs.entry(spend.run_id).or_default().spent = spend.spent; + } + } } // Keys are unique, so a run is dense from 1 exactly when its rows - // number its highest sequence. - if runs.iter().any(|(run, tally)| counts[run] != tally.last_sequence) { - return Err(CallRecordError::Corrupt); + // number its highest sequence. Every record was reserved before it was + // made, so the ledger covers what the records charged. + for (run, tally) in &runs { + let rows = counts.get(run).copied().unwrap_or(0); + let recorded = charged.get(run).copied().unwrap_or(0); + if rows != tally.last_sequence || tally.spent < recorded { + return Err(CallRecordError::Corrupt); + } } Ok(Self { - state: Mutex::new(StoreState { store, runs }), + state: Mutex::new(StoreState { + store, + runs, + owned: BTreeSet::new(), + }), }) } @@ -223,17 +309,90 @@ impl CallRecordStore { + 1) } - /// Tokens this run's stored calls were charged. - pub fn charged(&self, run: &RunId) -> Result { + /// Tokens this run has reserved or been charged: its persisted ledger. + pub fn spent(&self, run: &RunId) -> Result { Ok(self .lock()? .runs .get(run.as_str()) - .map_or(0, |tally| tally.charged)) + .map_or(0, |tally| tally.spent)) + } + + /// Takes the run's ledger for one port, or refuses when another holds it. + fn own(&self, run: &RunId) -> Result<(), CallRecordError> { + if self.lock()?.owned.insert(run.as_str().to_owned()) { + Ok(()) + } else { + Err(CallRecordError::RunInUse) + } + } + + fn disown(&self, run: &RunId) { + if let Ok(mut state) = self.state.lock() { + state.owned.remove(run.as_str()); + } + } + + /// Persists `spent` as the run's ledger, then adopts it in memory. A write + /// that fails leaves the previous figure. + fn write_spent(state: &mut StoreState, run: &str, spent: u64) -> Result<(), CallRecordError> { + let spend = RunSpend { + run_id: run.to_owned(), + spent, + }; + let row = CultCacheEnvelope { + key: run.to_owned(), + r#type: RUN_SPEND_ROW.into(), + payload: rmp_serde::to_vec_named(&spend).map_err(|_| CallRecordError::Store)?, + stored_at: Utc::now().to_rfc3339(), + schema_id: Some(RUN_SPEND_SCHEMA.into()), + }; + state.store.push(&row).map_err(|_| CallRecordError::Store)?; + state.runs.entry(run.to_owned()).or_default().spent = spent; + Ok(()) + } + + /// Charges `bound` to the run's persisted ledger, or refuses with + /// `OverCap` when the ledger plus the bound would pass `cap`. The check and + /// the write hold one lock, so calls on the wire hold their bounds. + fn reserve(&self, run: &RunId, bound: u64, cap: u64) -> Result<(), CallRecordError> { + let mut state = self.lock()?; + let held = state.runs.get(run.as_str()).map_or(0, |tally| tally.spent); + let after = held.saturating_add(bound); + if after > cap { + return Err(CallRecordError::OverCap); + } + Self::write_spent(&mut state, run.as_str(), after) + } + + /// Replaces a reservation of `bound` with what the call was charged. With + /// a record, the record is stored first, so a crash between the two writes + /// leaves the larger reservation charged. + fn settle( + &self, + run: &RunId, + bound: u64, + charged: u64, + record: Option<&ProviderCallRecord>, + ) -> Result<(), CallRecordError> { + let mut state = self.lock()?; + if let Some(record) = record { + Self::push_record(&mut state, record)?; + } + let held = state.runs.get(run.as_str()).map_or(0, |tally| tally.spent); + let settled = held.saturating_sub(bound).saturating_add(charged); + Self::write_spent(&mut state, run.as_str(), settled) } fn append(&self, record: &ProviderCallRecord) -> Result<(), CallRecordError> { let mut state = self.lock()?; + Self::push_record(&mut state, record) + } + + fn push_record( + state: &mut StoreState, + record: &ProviderCallRecord, + ) -> Result<(), CallRecordError> { let last = state .runs .get(&record.run_id) @@ -249,9 +408,11 @@ impl CallRecordStore { schema_id: Some(CALL_RECORD_SCHEMA.into()), }; state.store.push(&row).map_err(|_| CallRecordError::Store)?; - let tally = state.runs.entry(record.run_id.clone()).or_default(); - tally.last_sequence = record.sequence; - tally.charged = tally.charged.saturating_add(record.charged); + state + .runs + .entry(record.run_id.clone()) + .or_default() + .last_sequence = record.sequence; Ok(()) } @@ -265,54 +426,46 @@ impl CallRecordStore { .map_err(|_| CallRecordError::Store)?; let mut records = Vec::new(); for row in rows.iter().filter(|row| row.key.starts_with(&prefix)) { - records.push(decode_row(row)?); + if let DecodedRow::Call(record) = decode_row(row)? { + records.push(record); + } } records.sort_by_key(|record| record.sequence); Ok(records) } } -struct Ledger { - /// Tokens charged by finished calls, restored from the store at open. - settled: u64, - /// Bounds held by calls still on the wire. - in_flight: u64, -} - -/// Decorates a port for one run: refuses a call that could pass the cap, -/// charges the real cost, and stores every reply before returning it. +/// Decorates a port for one run: reserves each call's bound in the run's +/// ledger before the provider is called, settles it to the real cost, and +/// stores every reply before returning it. pub struct RecordingInferencePort { inner: Arc, store: Arc, run: RunId, cap: u64, - ledger: Mutex, } impl RecordingInferencePort { - /// The run's spend resumes from its stored records. + /// Takes ownership of the run's ledger, which resumes from the store. A + /// second port on a run whose first is alive is refused with `RunInUse`. pub fn open( inner: Arc, store: Arc, run: RunId, cap_tokens: u64, ) -> Result { - let settled = store.charged(&run)?; + store.own(&run)?; Ok(Self { inner, store, run, cap: cap_tokens, - ledger: Mutex::new(Ledger { - settled, - in_flight: 0, - }), }) } - /// Tokens charged to the run so far. + /// Tokens reserved or charged to the run so far. pub fn spent(&self) -> u64 { - self.ledger.lock().map_or(u64::MAX, |ledger| ledger.settled) + self.store.spent(&self.run).unwrap_or(u64::MAX) } /// The most this request can cost: its serialized bytes plus its output @@ -329,62 +482,49 @@ impl RecordingInferencePort { Ok((payload.len() as u64).saturating_add(u64::from(allowance))) } - fn reserve(&self, bound: u64) -> Result<(), InferenceFault> { - let mut ledger = self - .ledger - .lock() - .map_err(|_| InferenceFault::integrity_violation("the run's ledger was poisoned"))?; - let held = ledger.settled.saturating_add(ledger.in_flight); - if held.saturating_add(bound) > self.cap { - return Err(InferenceFault::budget_exhausted()); - } - ledger.in_flight += bound; - Ok(()) - } - fn settle( &self, bound: u64, identity: (InferencePurpose, String, String), result: Result, ) -> Result { - let mut ledger = self - .ledger - .lock() - .map_err(|_| InferenceFault::integrity_violation("the run's ledger was poisoned"))?; - ledger.in_flight = ledger.in_flight.saturating_sub(bound); - let output = match result { - Ok(output) => output, - Err(fault) => { - if fault.class() != InferenceFaultClass::Connect { - ledger.settled = ledger.settled.saturating_add(bound); - } - return Err(fault); - } - }; + // A fault leaves its reservation charged: the request may have run. + let output = result?; let charged = match output.usage { Some(usage) => usage.prompt.saturating_add(usage.completion), None => bound, }; - ledger.settled = ledger.settled.saturating_add(charged); - let (purpose, model, request_sha256) = identity; - let sequence = self.store.next_sequence(&self.run).map_err(unrecorded)?; + if charged > bound { + self.store + .settle(&self.run, bound, charged, None) + .map_err(unrecorded)?; + return Err(InferenceFault::budget_exhausted()); + } + let (purpose, model, content_sha256) = identity; let record = ProviderCallRecord { run_id: self.run.as_str().to_owned(), - sequence, + sequence: self.store.next_sequence(&self.run).map_err(unrecorded)?, purpose, model, - request_sha256, + content_sha256, events: output.events.clone(), receipt_digest: output.receipt_digest.clone(), usage: output.usage, charged, }; - self.store.append(&record).map_err(unrecorded)?; + self.store + .settle(&self.run, bound, charged, Some(&record)) + .map_err(unrecorded)?; Ok(output) } } +impl Drop for RecordingInferencePort { + fn drop(&mut self) { + self.store.disown(&self.run); + } +} + fn unrecorded(_: CallRecordError) -> InferenceFault { InferenceFault::integrity_violation("a provider call could not be recorded") } @@ -397,12 +537,17 @@ impl InferencePort for RecordingInferencePort { async fn infer(&self, request: PreparedInference) -> Result { let bound = Self::bound(&request)?; - self.reserve(bound)?; let identity = ( request.purpose, request.invocation.request.model.clone(), - hex(&request.invocation.provider_request_sha256), + content_digest(&request.invocation.request)?, ); + self.store + .reserve(&self.run, bound, self.cap) + .map_err(|error| match error { + CallRecordError::OverCap => InferenceFault::budget_exhausted(), + _ => unrecorded(error), + })?; let result = self.inner.infer(request).await; self.settle(bound, identity, result) } @@ -412,7 +557,8 @@ impl InferencePort for RecordingInferencePort { } } -/// Serves a run's stored replies, in sequence, to the same requests. It holds +/// Serves a run's stored replies, in sequence, to the same requests, matched +/// by what they ask of the provider and not by the ids that name them. It holds /// no provider and makes no call: a request that is not the next recorded one /// is a fault. pub struct ReplayInferencePort { @@ -454,13 +600,13 @@ impl InferencePort for ReplayInferencePort { .pending .lock() .map_err(|_| InferenceFault::integrity_violation("the replay queue was poisoned"))?; - let digest = hex(&request.invocation.provider_request_sha256); + let digest = content_digest(&request.invocation.request)?; let Some(next) = pending.front() else { return Err(InferenceFault::integrity_violation( "replay has no recorded call left", )); }; - if next.request_sha256 != digest { + if next.content_sha256 != digest { return Err(InferenceFault::integrity_violation( "the request is not the next recorded call", )); diff --git a/crates/ghostlight/src/verse/calls/tests.rs b/crates/ghostlight/src/verse/calls/tests.rs index eaecd1b..8de27c4 100644 --- a/crates/ghostlight/src/verse/calls/tests.rs +++ b/crates/ghostlight/src/verse/calls/tests.rs @@ -1,6 +1,8 @@ use super::*; use crate::CommandId; -use crate::controllers::{InferenceFaultDisposition, RequestShape, tool_request}; +use crate::controllers::{ + InferenceFaultClass, InferenceFaultDisposition, RequestShape, tool_request, +}; use codex_connector::{CodexInputItem, CodexToolDefinition}; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -171,7 +173,7 @@ async fn recording_port_stores_each_call_before_returning() { let prepared = port .prepare(request_for(CommandId::new(), "tell", 1_200)) .unwrap(); - let digest = hex(&prepared.invocation.provider_request_sha256); + let digest = content_digest(&prepared.invocation.request).unwrap(); let returned = port.infer(prepared).await.unwrap(); assert_eq!(json(&returned), json(expected)); let stored = store.records(&run).unwrap(); @@ -183,7 +185,7 @@ async fn recording_port_stores_each_call_before_returning() { let record = &stored[index]; assert_eq!(record.sequence(), index as u64 + 1); assert_eq!(record.run_id(), "run-a"); - assert_eq!(record.request_sha256(), digest); + assert_eq!(record.content_sha256(), digest); assert_eq!(record.model(), MODEL); assert_eq!(record.purpose(), InferencePurpose::Persona); assert_eq!(record.usage(), expected.usage); @@ -251,6 +253,7 @@ async fn cap_refuses_before_the_provider_is_called() { .is_empty() ); assert_eq!(port.spent(), 0); + drop(port); // Exactly the cap is allowed. let exact = recording(&inner, &store, "run-cap", bound); @@ -389,7 +392,7 @@ async fn a_reply_with_no_usage_is_charged_its_bound() { } #[tokio::test] -async fn a_fault_is_charged_its_bound_unless_it_never_connected() { +async fn a_fault_keeps_its_reservation_charged() { let directory = tempfile::tempdir().unwrap(); let (store, _) = store_in(&directory); let prepared = prepared_for("fault", 1_200); @@ -397,23 +400,21 @@ async fn a_fault_is_charged_its_bound_unless_it_never_connected() { let inner = ScriptedPort::new(vec![ Err(InferenceFault::retryable("down").classed(InferenceFaultClass::Connect)), Err(InferenceFault::retryable("slow").classed(InferenceFaultClass::Timeout)), + Err(InferenceFault::new("junk").classed(InferenceFaultClass::BadReply)), ]); let port = recording(&inner, &store, "run-fault", u64::MAX); - port.infer(prepared.clone()) - .await - .expect_err("connect fault"); - assert_eq!(port.spent(), 0, "nothing was sent"); - let fault = port.infer(prepared).await.expect_err("timeout fault"); - assert_eq!( - fault.class(), + for (index, class) in [ + InferenceFaultClass::Connect, InferenceFaultClass::Timeout, - "the fault passes through" - ); - assert_eq!( - port.spent(), - bound, - "a request that may have run is charged its bound" - ); + InferenceFaultClass::BadReply, + ] + .into_iter() + .enumerate() + { + let fault = port.infer(prepared.clone()).await.expect_err("a fault"); + assert_eq!(fault.class(), class, "the fault passes through"); + assert_eq!(port.spent(), bound * (index as u64 + 1)); + } assert!( store .records(&RunId::new("run-fault").unwrap()) @@ -422,6 +423,151 @@ async fn a_fault_is_charged_its_bound_unless_it_never_connected() { ); } +#[tokio::test] +async fn faulting_calls_across_restarts_never_pass_the_cap() { + let directory = tempfile::tempdir().unwrap(); + let (store, path) = store_in(&directory); + let prepared = prepared_for("loop", 1_200); + let bound = bound_of(&prepared); + let cap = bound * 3 + bound / 2; + drop(store); + let mut reached_the_provider = 0; + for attempt in 0..12 { + let store = Arc::new(CallRecordStore::open(&path).unwrap()); + let class = if attempt % 2 == 0 { + InferenceFaultClass::BadReply + } else { + InferenceFaultClass::Timeout + }; + let inner = ScriptedPort::new(vec![Err(InferenceFault::new("down").classed(class))]); + let port = recording(&inner, &store, "run-loop", cap); + let fault = port.infer(prepared.clone()).await.expect_err("a fault"); + reached_the_provider += inner.calls(); + if inner.calls() == 0 { + assert_eq!(fault.class(), InferenceFaultClass::BudgetExhausted); + } + assert!(port.spent() <= cap, "{} > {cap}", port.spent()); + } + assert_eq!(reached_the_provider, 3, "a restart returns no budget"); +} + +#[tokio::test] +async fn a_call_abandoned_on_the_wire_stays_charged_after_a_restart() { + let directory = tempfile::tempdir().unwrap(); + let (store, path) = store_in(&directory); + let prepared = prepared_for("crash", 1_200); + let bound = bound_of(&prepared); + let gate = Arc::new(tokio::sync::Semaphore::new(0)); + let inner = ScriptedPort::gated(vec![Ok(reply("never", "r"))], gate); + let port = Arc::new(recording(&inner, &store, "run-crash", bound)); + let call = { + let port = port.clone(); + let prepared = prepared.clone(); + tokio::spawn(async move { port.infer(prepared).await }) + }; + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while inner.calls() == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("the call reached the provider"); + call.abort(); + let _ = call.await; + drop((port, store)); + + let store = Arc::new(CallRecordStore::open(&path).unwrap()); + let inner = ScriptedPort::new(vec![Ok(reply("again", "r2"))]); + let reopened = recording(&inner, &store, "run-crash", bound); + assert_eq!(reopened.spent(), bound); + let fault = reopened + .infer(prepared) + .await + .expect_err("the cap is spent"); + assert_eq!(fault.class(), InferenceFaultClass::BudgetExhausted); + assert_eq!(inner.calls(), 0); +} + +#[tokio::test] +async fn one_run_has_one_ledger_and_one_owner() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let prepared = prepared_for("owner", 1_200); + let bound = bound_of(&prepared); + let cap = bound + bound / 2; + let inner = ScriptedPort::new(vec![Ok(reply("a", "r1")), Ok(reply("b", "r2"))]); + let first = recording(&inner, &store, "run-own", cap); + let second = RecordingInferencePort::open( + inner.clone(), + store.clone(), + RunId::new("run-own").unwrap(), + cap, + ); + assert_eq!(second.err(), Some(CallRecordError::RunInUse)); + // Another run is another ledger. + drop(recording(&inner, &store, "run-other", cap)); + first.infer(prepared.clone()).await.unwrap(); + assert_eq!(first.spent(), bound); + // The owner going away frees the run, and its spend carries over. + drop(first); + let next = recording(&inner, &store, "run-own", cap); + assert_eq!(next.spent(), bound); + next.infer(prepared) + .await + .expect_err("the cap holds across owners"); + assert_eq!(inner.calls(), 1); +} + +#[tokio::test] +async fn usage_over_the_bound_is_charged_in_full_and_faults_the_call() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let prepared = prepared_for("liar", 1_200); + let bound = bound_of(&prepared); + let cap = bound + 200; + let run = RunId::new("run-over").unwrap(); + let inner = ScriptedPort::new(vec![ + Ok(reply("big", "r1").with_usage(usage(bound, bound, None))), + Ok(reply("never", "r2")), + ]); + let port = recording(&inner, &store, "run-over", cap); + let fault = port.infer(prepared.clone()).await.expect_err("over bound"); + assert_eq!(fault.class(), InferenceFaultClass::BudgetExhausted); + assert_eq!( + fault.disposition(), + InferenceFaultDisposition::IntegrityViolation + ); + assert_eq!(port.spent(), bound * 2, "charged in full"); + assert!(store.records(&run).unwrap().is_empty()); + // The run is stopped: the next call is refused before the provider. + let fault = port.infer(prepared).await.expect_err("the cap is passed"); + assert_eq!(fault.class(), InferenceFaultClass::BudgetExhausted); + assert_eq!(inner.calls(), 1); +} + +#[tokio::test] +async fn usage_exactly_at_the_bound_is_a_normal_reply() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let prepared = prepared_for("exact", 1_200); + let bound = bound_of(&prepared); + let inner = ScriptedPort::new(vec![Ok(reply("fits", "r1").with_usage(usage( + bound - 10, + 10, + None, + )))]); + let port = recording(&inner, &store, "run-exact", bound); + port.infer(prepared).await.expect("a reply at its bound"); + assert_eq!(port.spent(), bound); + assert_eq!( + store + .records(&RunId::new("run-exact").unwrap()) + .unwrap() + .len(), + 1 + ); +} + #[tokio::test] async fn a_reopened_run_resumes_its_spend() { let directory = tempfile::tempdir().unwrap(); @@ -477,7 +623,7 @@ async fn replay_serves_stored_responses_and_refuses_strangers() { let replay = ReplayInferencePort::new("verse-replay", stored).unwrap(); // A request the run never made is a fault and consumes nothing. let stranger = replay - .prepare(request_for(CommandId::new(), "a", 1_200)) + .prepare(request_for(CommandId::new(), "z", 1_200)) .unwrap(); let fault = replay.infer(stranger).await.expect_err("a stranger"); assert_eq!( @@ -488,10 +634,7 @@ async fn replay_serves_stored_responses_and_refuses_strangers() { let second_first = replay .prepare(request_for(commands[1], "b", 1_200)) .unwrap(); - replay - .infer(second_first) - .await - .expect_err("out of order"); + replay.infer(second_first).await.expect_err("out of order"); // In order, each request gets exactly what the run received. for ((command, prompt), expected) in commands.iter().zip(["a", "b", "c"]).zip(&outputs) { let prepared = replay @@ -513,7 +656,7 @@ fn sample_record(run: &str, sequence: u64) -> ProviderCallRecord { sequence, purpose: InferencePurpose::Persona, model: MODEL.into(), - request_sha256: "ab".into(), + content_sha256: "ab".into(), events: vec![InferenceEvent::Text("t".into())], receipt_digest: "r".into(), usage: None, @@ -551,7 +694,7 @@ async fn a_record_holds_no_request_text() { .unwrap() .pull_all() .unwrap(); - assert_eq!(raw.len(), 1); + assert_eq!(raw.len(), 2, "one record and the run's ledger"); for row in raw { let bytes = [row.key.as_bytes(), row.payload.as_slice()].concat(); assert!( @@ -565,6 +708,19 @@ async fn a_record_holds_no_request_text() { assert!(!format!("{refusal} {refusal:?}").contains(canary)); } +fn spend_row(run: &str, spent: u64) -> CultCacheEnvelope { + envelope( + RUN_SPEND_ROW, + Some(RUN_SPEND_SCHEMA), + run, + rmp_serde::to_vec_named(&RunSpend { + run_id: run.into(), + spent, + }) + .unwrap(), + ) +} + fn envelope( row_type: &str, schema: Option<&str>, @@ -640,6 +796,37 @@ fn the_store_refuses_any_row_that_is_not_a_canonical_v1_record() { ), ("sequence zero", record_at(0)), ("a hole in the run", record_at(2)), + ("a record with no ledger", record_at(1)), + ( + "a ledger at the wrong key", + envelope( + RUN_SPEND_ROW, + Some(RUN_SPEND_SCHEMA), + "other", + rmp_serde::to_vec_named(&RunSpend { + run_id: "run-x".into(), + spent: 1, + }) + .unwrap(), + ), + ), + ( + "a ledger that is not canonical", + envelope( + RUN_SPEND_ROW, + Some(RUN_SPEND_SCHEMA), + "run-x", + [ + rmp_serde::to_vec_named(&RunSpend { + run_id: "run-x".into(), + spent: 1, + }) + .unwrap(), + vec![0xc0], + ] + .concat(), + ), + ), ]; for (label, row) in cases { let directory = tempfile::tempdir().unwrap(); @@ -651,17 +838,28 @@ fn the_store_refuses_any_row_that_is_not_a_canonical_v1_record() { let error = CallRecordStore::open(&path).err(); assert_eq!(error, Some(CallRecordError::Corrupt), "{label}"); } - // The same row, well formed, opens. + // A ledger under the charges of its records is refused too. let directory = tempfile::tempdir().unwrap(); let path = directory.path().join("calls.redb"); - OwnedRedbMessagePackBackingStore::new(&path) - .unwrap() - .push(&v1(&key, payload)) - .unwrap(); + let mut backing = OwnedRedbMessagePackBackingStore::new(&path).unwrap(); + backing.push(&v1(&key, payload.clone())).unwrap(); + backing.push(&spend_row("run-x", 2)).unwrap(); + drop(backing); + assert_eq!( + CallRecordStore::open(&path).err(), + Some(CallRecordError::Corrupt) + ); + // The same rows, well formed and covered by the ledger, open. + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("calls.redb"); + let mut backing = OwnedRedbMessagePackBackingStore::new(&path).unwrap(); + backing.push(&v1(&key, payload)).unwrap(); + backing.push(&spend_row("run-x", 3)).unwrap(); + drop(backing); let store = CallRecordStore::open(&path).unwrap(); let run = RunId::new("run-x").unwrap(); assert_eq!(store.records(&run).unwrap(), vec![good]); - assert_eq!(store.charged(&run).unwrap(), 3); + assert_eq!(store.spent(&run).unwrap(), 3); } #[test] @@ -694,3 +892,39 @@ fn a_run_id_is_a_key_safe_segment() { assert_eq!(RunId::new(bad), Err(CallRecordError::InvalidRunId)); } } + +#[tokio::test] +async fn replay_keys_the_request_content_not_the_command_identity() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let run = RunId::new("run-world").unwrap(); + let inner = ScriptedPort::new(ok_all(&[reply("one", "r1"), reply("two", "r2")])); + let live = recording(&inner, &store, "run-world", u64::MAX); + for prompt in ["first", "second"] { + live.infer( + live.prepare(request_for(CommandId::new(), prompt, 1_200)) + .unwrap(), + ) + .await + .unwrap(); + } + // A fresh genesis names every command anew; the content is what repeats. + let replay = ReplayInferencePort::new("verse-replay", store.records(&run).unwrap()).unwrap(); + let under_new_names = |prompt: &str, max: u32| { + replay + .prepare(request_for(CommandId::new(), prompt, max)) + .unwrap() + }; + // The same words under a different output allowance are another request. + replay + .infer(under_new_names("first", 1_201)) + .await + .expect_err("a different shape"); + let one = replay.infer(under_new_names("first", 1_200)).await.unwrap(); + let two = replay + .infer(under_new_names("second", 1_200)) + .await + .unwrap(); + assert_eq!(json(&one), json(&reply("one", "r1"))); + assert_eq!(json(&two), json(&reply("two", "r2"))); +} diff --git a/crates/ghostlight/src/verse/mod.rs b/crates/ghostlight/src/verse/mod.rs index 5c1e308..7c7b6fb 100644 --- a/crates/ghostlight/src/verse/mod.rs +++ b/crates/ghostlight/src/verse/mod.rs @@ -5,6 +5,6 @@ mod calls; pub use calls::{ - CALL_RECORD_SCHEMA, CallRecordError, CallRecordStore, ProviderCallRecord, RecordingInferencePort, - ReplayInferencePort, RunId, + CALL_RECORD_SCHEMA, CallRecordError, CallRecordStore, ProviderCallRecord, + RecordingInferencePort, ReplayInferencePort, RunId, }; From 8c9b88b519338aa6f68d8f7f4976052e304219f9 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 10 Oct 2026 17:22:02 +0200 Subject: [PATCH 06/10] call-record: drop two record-decode checks the ledger and density rules already make A record with a run id that is no key segment has no readable ledger, and a record at sequence zero fails the density rule at open; the two decode_row conditions were equivalent mutants (cargo-mutants missed both). Builds. Co-Authored-By: Claude Opus 5.5 --- crates/ghostlight/src/verse/calls.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/ghostlight/src/verse/calls.rs b/crates/ghostlight/src/verse/calls.rs index cfd3460..dbf7e9d 100644 --- a/crates/ghostlight/src/verse/calls.rs +++ b/crates/ghostlight/src/verse/calls.rs @@ -207,10 +207,10 @@ fn decode_row(row: &CultCacheEnvelope) -> Result { match (row.r#type.as_str(), row.schema_id.as_deref()) { (CALL_RECORD_ROW, Some(CALL_RECORD_SCHEMA)) => { let record: ProviderCallRecord = canonical(row)?; - if RunId::new(&record.run_id).is_err() - || record.sequence == 0 - || row.key != row_key(&record.run_id, record.sequence) - { + // The run id and a sequence of zero need no check here: a run id + // that is no key segment has no readable ledger, and a zero + // sequence fails the density rule in `open`. + if row.key != row_key(&record.run_id, record.sequence) { return Err(CallRecordError::Corrupt); } Ok(DecodedRow::Call(record)) From 2c5eb95da3f1b0d1d9a2fdc490fe04fa96205b7e Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 10 Oct 2026 20:32:17 +0200 Subject: [PATCH 07/10] Call record: ledger carries cap and starting world; replay refuses another world The run ledger is the one owner of what a run began as: its cap and the world id and state digest it started from. Replay takes the store, the run and a restored world and refuses a different world up front with a typed error naming the field. Restores the decode checks the ledger rules do not imply, makes the replay key an exhaustive destructure, and pins the over-bound edge. Not yet built or tested at this commit. Co-Authored-By: Claude Opus 5.5 --- crates/ghostlight/src/lib.rs | 2 +- crates/ghostlight/src/verse/calls.rs | 264 ++++++++++++----- crates/ghostlight/src/verse/calls/tests.rs | 330 +++++++++++++++++++-- crates/ghostlight/src/verse/mod.rs | 2 +- 4 files changed, 499 insertions(+), 99 deletions(-) diff --git a/crates/ghostlight/src/lib.rs b/crates/ghostlight/src/lib.rs index 409a251..ce1ba8f 100644 --- a/crates/ghostlight/src/lib.rs +++ b/crates/ghostlight/src/lib.rs @@ -133,7 +133,7 @@ pub use table::{ pub use vault::{VaultError, VaultEvidenceSource}; pub use verse::{ CALL_RECORD_SCHEMA, CallRecordError, CallRecordStore, ProviderCallRecord, RecordingInferencePort, - ReplayInferencePort, RunId, + ReplayInferencePort, RunField, RunId, RunStart, }; use patch::{EdgeRecord, EntityRecord, LedgerDelta, ResolvedOp, ResolvedPatch}; diff --git a/crates/ghostlight/src/verse/calls.rs b/crates/ghostlight/src/verse/calls.rs index dbf7e9d..4392114 100644 --- a/crates/ghostlight/src/verse/calls.rs +++ b/crates/ghostlight/src/verse/calls.rs @@ -38,9 +38,9 @@ use std::{ use thiserror::Error; const CALL_RECORD_ROW: &str = "verse_provider_call.v1"; -const RUN_SPEND_ROW: &str = "verse_run_spend.v1"; +const RUN_LEDGER_ROW: &str = "verse_run_ledger.v1"; /// The run ledger document's schema. -const RUN_SPEND_SCHEMA: &str = "ghostlight.verse_run_spend.v1"; +const RUN_LEDGER_SCHEMA: &str = "ghostlight.verse_run_ledger.v1"; /// The call record document's schema. A store holding any other row type or /// schema is refused at open: this store is not migrated from, or shared with, /// another document. @@ -62,6 +62,46 @@ pub enum CallRecordError { RunInUse, #[error("the call could pass the run's token cap")] OverCap, + #[error("the run was recorded with a different {0}")] + RunMismatch(RunField), +} + +/// What a run is bound to when it begins, named by the field a mismatch refuses. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum RunField { + /// The token cap the run was opened with. + Cap, + /// The identity of the world the run started from. + WorldId, + /// The state digest of the world the run started from. + StateDigest, +} + +impl std::fmt::Display for RunField { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + Self::Cap => "token cap", + Self::WorldId => "world id", + Self::StateDigest => "world state digest", + }) + } +} + +/// The world a run began from: its identity and state digest. Replay is of a +/// restored copy of this world, so it takes the same two values. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RunStart { + world_id: String, + state_digest: String, +} + +impl RunStart { + pub fn of(world: &crate::WorldSnapshot) -> Self { + Self { + world_id: world.world_id.text(), + state_digest: world.state_digest.clone(), + } + } } /// A run's identity as a key segment. @@ -158,37 +198,62 @@ fn hex(digest: &[u8]) -> String { /// one world's random identity. Everything else the provider is sent /// (model, instructions, input, tools, shape) is in the digest. fn content_digest(request: &CodexProviderRequest) -> Result { + // Exhaustive on purpose: a field added upstream must be placed in the key + // or left out here by name before this compiles. + let CodexProviderRequest { + schema_id, + request_id: _request_id, // names the call, not what it asks + conversation_id: _conversation_id, // names the call, not what it asks + model, + instructions, + input, + reasoning_effort, + reasoning_summary, + service_tier, + output_format_name, + previous_response_id, + tools, + tool_choice, + parallel_tool_calls, + output_schema_json, + max_output_tokens, + prompt_cache_key, + } = request; let bytes = rmp_serde::to_vec(&( - &request.schema_id, - &request.model, - &request.instructions, - &request.input, - &request.reasoning_effort, - &request.reasoning_summary, - &request.service_tier, - &request.output_format_name, - &request.previous_response_id, - &request.tools, - &request.tool_choice, - request.parallel_tool_calls, - &request.output_schema_json, - request.max_output_tokens, - &request.prompt_cache_key, + schema_id, + model, + instructions, + input, + reasoning_effort, + reasoning_summary, + service_tier, + output_format_name, + previous_response_id, + tools, + tool_choice, + parallel_tool_calls, + output_schema_json, + max_output_tokens, + prompt_cache_key, )) .map_err(|_| InferenceFault::new("a provider request could not be measured"))?; Ok(hex(&Sha256::digest(bytes))) } -/// A run's persisted ledger: tokens reserved or charged against its cap. +/// A run's persisted ledger, the one owner of what the run began as: the cap +/// it was opened under, the world it started from, and the tokens reserved or +/// charged so far. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -struct RunSpend { +struct RunLedger { run_id: String, + cap: u64, + start: RunStart, spent: u64, } enum DecodedRow { Call(ProviderCallRecord), - Spend(RunSpend), + Ledger(RunLedger), } fn canonical Deserialize<'de>>( @@ -207,20 +272,22 @@ fn decode_row(row: &CultCacheEnvelope) -> Result { match (row.r#type.as_str(), row.schema_id.as_deref()) { (CALL_RECORD_ROW, Some(CALL_RECORD_SCHEMA)) => { let record: ProviderCallRecord = canonical(row)?; - // The run id and a sequence of zero need no check here: a run id - // that is no key segment has no readable ledger, and a zero - // sequence fails the density rule in `open`. - if row.key != row_key(&record.run_id, record.sequence) { + // A run id that is no whole key segment would key into another + // run's records, and sequences count from 1. + if RunId::new(&record.run_id).is_err() + || record.sequence == 0 + || row.key != row_key(&record.run_id, record.sequence) + { return Err(CallRecordError::Corrupt); } Ok(DecodedRow::Call(record)) } - (RUN_SPEND_ROW, Some(RUN_SPEND_SCHEMA)) => { - let spend: RunSpend = canonical(row)?; - if RunId::new(&spend.run_id).is_err() || row.key != spend.run_id { + (RUN_LEDGER_ROW, Some(RUN_LEDGER_SCHEMA)) => { + let ledger: RunLedger = canonical(row)?; + if RunId::new(&ledger.run_id).is_err() || row.key != ledger.run_id { return Err(CallRecordError::Corrupt); } - Ok(DecodedRow::Spend(spend)) + Ok(DecodedRow::Ledger(ledger)) } _ => Err(CallRecordError::Corrupt), } @@ -229,8 +296,8 @@ fn decode_row(row: &CultCacheEnvelope) -> Result { #[derive(Default)] struct RunTally { last_sequence: u64, - /// Tokens reserved or charged: the persisted ledger, read at open. - spent: u64, + /// The persisted ledger, read at open and written where it is spent. + ledger: Option, } struct StoreState { @@ -271,18 +338,23 @@ impl CallRecordStore { let tally = runs.entry(record.run_id).or_default(); tally.last_sequence = tally.last_sequence.max(record.sequence); } - DecodedRow::Spend(spend) => { - runs.entry(spend.run_id).or_default().spent = spend.spent; + DecodedRow::Ledger(ledger) => { + runs.entry(ledger.run_id.clone()).or_default().ledger = Some(ledger); } } } - // Keys are unique, so a run is dense from 1 exactly when its rows - // number its highest sequence. Every record was reserved before it was - // made, so the ledger covers what the records charged. + // Sequences start at 1 and keys are unique, so a run is dense exactly + // when its rows number its highest sequence. Every record was reserved + // in a ledger before it was made, so a run with records has a ledger + // that covers what they charged. for (run, tally) in &runs { let rows = counts.get(run).copied().unwrap_or(0); let recorded = charged.get(run).copied().unwrap_or(0); - if rows != tally.last_sequence || tally.spent < recorded { + let covered = tally + .ledger + .as_ref() + .is_some_and(|ledger| ledger.spent >= recorded); + if rows != tally.last_sequence || !covered { return Err(CallRecordError::Corrupt); } } @@ -315,16 +387,46 @@ impl CallRecordStore { .lock()? .runs .get(run.as_str()) - .map_or(0, |tally| tally.spent)) + .and_then(|tally| tally.ledger.as_ref()) + .map_or(0, |ledger| ledger.spent)) } /// Takes the run's ledger for one port, or refuses when another holds it. - fn own(&self, run: &RunId) -> Result<(), CallRecordError> { - if self.lock()?.owned.insert(run.as_str().to_owned()) { - Ok(()) - } else { - Err(CallRecordError::RunInUse) + /// The first claim of a run writes its ledger: the cap it is opened under + /// and the world it starts from. A later claim must name the same cap; the + /// start is not compared, because a resumed world has moved on. + fn claim(&self, run: &RunId, cap: u64, start: &RunStart) -> Result<(), CallRecordError> { + let mut state = self.lock()?; + if !state.owned.insert(run.as_str().to_owned()) { + return Err(CallRecordError::RunInUse); } + let begun = match state.runs.get(run.as_str()).and_then(|t| t.ledger.as_ref()) { + Some(ledger) if ledger.cap == cap => Ok(()), + Some(_) => Err(CallRecordError::RunMismatch(RunField::Cap)), + None => Self::write_ledger( + &mut state, + RunLedger { + run_id: run.as_str().to_owned(), + cap, + start: start.clone(), + spent: 0, + }, + ), + }; + if begun.is_err() { + state.owned.remove(run.as_str()); + } + begun + } + + /// The world a run began from, when it has begun. + fn start_of(&self, run: &RunId) -> Result, CallRecordError> { + Ok(self + .lock()? + .runs + .get(run.as_str()) + .and_then(|tally| tally.ledger.as_ref()) + .map(|ledger| ledger.start.clone())) } fn disown(&self, run: &RunId) { @@ -333,32 +435,46 @@ impl CallRecordStore { } } - /// Persists `spent` as the run's ledger, then adopts it in memory. A write - /// that fails leaves the previous figure. - fn write_spent(state: &mut StoreState, run: &str, spent: u64) -> Result<(), CallRecordError> { - let spend = RunSpend { - run_id: run.to_owned(), - spent, - }; + /// Persists the ledger, then adopts it in memory. A write that fails + /// leaves the previous ledger. + fn write_ledger(state: &mut StoreState, ledger: RunLedger) -> Result<(), CallRecordError> { let row = CultCacheEnvelope { - key: run.to_owned(), - r#type: RUN_SPEND_ROW.into(), - payload: rmp_serde::to_vec_named(&spend).map_err(|_| CallRecordError::Store)?, + key: ledger.run_id.clone(), + r#type: RUN_LEDGER_ROW.into(), + payload: rmp_serde::to_vec_named(&ledger).map_err(|_| CallRecordError::Store)?, stored_at: Utc::now().to_rfc3339(), - schema_id: Some(RUN_SPEND_SCHEMA.into()), + schema_id: Some(RUN_LEDGER_SCHEMA.into()), }; state.store.push(&row).map_err(|_| CallRecordError::Store)?; - state.runs.entry(run.to_owned()).or_default().spent = spent; + state.runs.entry(ledger.run_id.clone()).or_default().ledger = Some(ledger); Ok(()) } + /// Persists `spent` as the run's spend; the run must have been claimed. + fn write_spent(state: &mut StoreState, run: &str, spent: u64) -> Result<(), CallRecordError> { + let mut ledger = state + .runs + .get(run) + .and_then(|tally| tally.ledger.clone()) + .ok_or(CallRecordError::Store)?; + ledger.spent = spent; + Self::write_ledger(state, ledger) + } + + fn held(state: &StoreState, run: &RunId) -> u64 { + state + .runs + .get(run.as_str()) + .and_then(|tally| tally.ledger.as_ref()) + .map_or(0, |ledger| ledger.spent) + } + /// Charges `bound` to the run's persisted ledger, or refuses with /// `OverCap` when the ledger plus the bound would pass `cap`. The check and /// the write hold one lock, so calls on the wire hold their bounds. fn reserve(&self, run: &RunId, bound: u64, cap: u64) -> Result<(), CallRecordError> { let mut state = self.lock()?; - let held = state.runs.get(run.as_str()).map_or(0, |tally| tally.spent); - let after = held.saturating_add(bound); + let after = Self::held(&state, run).saturating_add(bound); if after > cap { return Err(CallRecordError::OverCap); } @@ -379,8 +495,9 @@ impl CallRecordStore { if let Some(record) = record { Self::push_record(&mut state, record)?; } - let held = state.runs.get(run.as_str()).map_or(0, |tally| tally.spent); - let settled = held.saturating_sub(bound).saturating_add(charged); + let settled = Self::held(&state, run) + .saturating_sub(bound) + .saturating_add(charged); Self::write_spent(&mut state, run.as_str(), settled) } @@ -446,15 +563,18 @@ pub struct RecordingInferencePort { } impl RecordingInferencePort { - /// Takes ownership of the run's ledger, which resumes from the store. A - /// second port on a run whose first is alive is refused with `RunInUse`. + /// Takes ownership of the run's ledger, which resumes from the store. The + /// first open of a run records its cap and the world it starts from; a + /// reopen with another cap is refused with `RunMismatch(Cap)`. A second + /// port on a run whose first is alive is refused with `RunInUse`. pub fn open( inner: Arc, store: Arc, run: RunId, cap_tokens: u64, + start: &RunStart, ) -> Result { - store.own(&run)?; + store.claim(&run, cap_tokens, start)?; Ok(Self { inner, store, @@ -567,20 +687,26 @@ pub struct ReplayInferencePort { } impl ReplayInferencePort { - /// `records` are one run's, as `CallRecordStore::records` returns them. + /// Replays `run` from the store against `world`, a restored copy of the + /// world the run started from. Any other world is refused here, before a + /// call, with `RunMismatch` naming the field; a run the store never began + /// is `NotReplayable`. pub fn new( caller_runtime_id: impl Into, - records: Vec, + store: &CallRecordStore, + run: &RunId, + world: &RunStart, ) -> Result { - let dense = records.iter().enumerate().all(|(index, record)| { - record.sequence == index as u64 + 1 && record.run_id == records[0].run_id - }); - if !dense { - return Err(CallRecordError::NotReplayable); + let began = store.start_of(run)?.ok_or(CallRecordError::NotReplayable)?; + if began.world_id != world.world_id { + return Err(CallRecordError::RunMismatch(RunField::WorldId)); + } + if began.state_digest != world.state_digest { + return Err(CallRecordError::RunMismatch(RunField::StateDigest)); } Ok(Self { caller_runtime_id: caller_runtime_id.into(), - pending: Mutex::new(records.into()), + pending: Mutex::new(store.records(run)?.into()), }) } } diff --git a/crates/ghostlight/src/verse/calls/tests.rs b/crates/ghostlight/src/verse/calls/tests.rs index 8de27c4..ca88565 100644 --- a/crates/ghostlight/src/verse/calls/tests.rs +++ b/crates/ghostlight/src/verse/calls/tests.rs @@ -129,14 +129,28 @@ fn store_in(directory: &tempfile::TempDir) -> (Arc, std::path:: ) } +/// A world identity and digest for tests that need a run to start somewhere. +fn start() -> RunStart { + RunStart { + world_id: "world-a".into(), + state_digest: "sha256:a".into(), + } +} + fn recording( inner: &Arc, store: &Arc, run: &str, cap: u64, ) -> RecordingInferencePort { - RecordingInferencePort::open(inner.clone(), store.clone(), RunId::new(run).unwrap(), cap) - .expect("the port opens") + RecordingInferencePort::open( + inner.clone(), + store.clone(), + RunId::new(run).unwrap(), + cap, + &start(), + ) + .expect("the port opens") } fn json(output: &InferenceOutput) -> Vec { @@ -502,6 +516,7 @@ async fn one_run_has_one_ledger_and_one_owner() { store.clone(), RunId::new("run-own").unwrap(), cap, + &start(), ); assert_eq!(second.err(), Some(CallRecordError::RunInUse)); // Another run is another ledger. @@ -583,7 +598,7 @@ async fn a_reopened_run_resumes_its_spend() { let store = Arc::new(CallRecordStore::open(&path).unwrap()); let inner = ScriptedPort::new(vec![Ok(reply("two", "r2"))]); - let resumed = recording(&inner, &store, "run-resume", bound * 2 - 1); + let resumed = recording(&inner, &store, "run-resume", bound * 2); assert_eq!(resumed.spent(), bound); let fault = resumed .infer(probe) @@ -619,8 +634,7 @@ async fn replay_serves_stored_responses_and_refuses_strangers() { } let calls_made = inner.calls(); - let stored = store.records(&run).unwrap(); - let replay = ReplayInferencePort::new("verse-replay", stored).unwrap(); + let replay = ReplayInferencePort::new("verse-replay", &store, &run, &start()).unwrap(); // A request the run never made is a fault and consumes nothing. let stranger = replay .prepare(request_for(CommandId::new(), "z", 1_200)) @@ -665,20 +679,14 @@ fn sample_record(run: &str, sequence: u64) -> ProviderCallRecord { } #[test] -fn replay_needs_one_runs_dense_records() { - assert!( - ReplayInferencePort::new("x", vec![sample_record("a", 1), sample_record("a", 2)]).is_ok() +fn replay_needs_a_run_the_store_began() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let run = RunId::new("never-begun").unwrap(); + assert_eq!( + ReplayInferencePort::new("x", &store, &run, &start()).err(), + Some(CallRecordError::NotReplayable) ); - for records in [ - vec![sample_record("a", 2)], - vec![sample_record("a", 1), sample_record("a", 3)], - vec![sample_record("a", 1), sample_record("b", 2)], - ] { - assert_eq!( - ReplayInferencePort::new("x", records).err(), - Some(CallRecordError::NotReplayable) - ); - } } #[tokio::test] @@ -710,11 +718,13 @@ async fn a_record_holds_no_request_text() { fn spend_row(run: &str, spent: u64) -> CultCacheEnvelope { envelope( - RUN_SPEND_ROW, - Some(RUN_SPEND_SCHEMA), + RUN_LEDGER_ROW, + Some(RUN_LEDGER_SCHEMA), run, - rmp_serde::to_vec_named(&RunSpend { + rmp_serde::to_vec_named(&RunLedger { run_id: run.into(), + cap: u64::MAX, + start: start(), spent, }) .unwrap(), @@ -800,11 +810,13 @@ fn the_store_refuses_any_row_that_is_not_a_canonical_v1_record() { ( "a ledger at the wrong key", envelope( - RUN_SPEND_ROW, - Some(RUN_SPEND_SCHEMA), + RUN_LEDGER_ROW, + Some(RUN_LEDGER_SCHEMA), "other", - rmp_serde::to_vec_named(&RunSpend { + rmp_serde::to_vec_named(&RunLedger { run_id: "run-x".into(), + cap: u64::MAX, + start: start(), spent: 1, }) .unwrap(), @@ -813,12 +825,14 @@ fn the_store_refuses_any_row_that_is_not_a_canonical_v1_record() { ( "a ledger that is not canonical", envelope( - RUN_SPEND_ROW, - Some(RUN_SPEND_SCHEMA), + RUN_LEDGER_ROW, + Some(RUN_LEDGER_SCHEMA), "run-x", [ - rmp_serde::to_vec_named(&RunSpend { + rmp_serde::to_vec_named(&RunLedger { run_id: "run-x".into(), + cap: u64::MAX, + start: start(), spent: 1, }) .unwrap(), @@ -909,7 +923,7 @@ async fn replay_keys_the_request_content_not_the_command_identity() { .unwrap(); } // A fresh genesis names every command anew; the content is what repeats. - let replay = ReplayInferencePort::new("verse-replay", store.records(&run).unwrap()).unwrap(); + let replay = ReplayInferencePort::new("verse-replay", &store, &run, &start()).unwrap(); let under_new_names = |prompt: &str, max: u32| { replay .prepare(request_for(CommandId::new(), prompt, max)) @@ -928,3 +942,263 @@ async fn replay_keys_the_request_content_not_the_command_identity() { assert_eq!(json(&one), json(&reply("one", "r1"))); assert_eq!(json(&two), json(&reply("two", "r2"))); } + +#[tokio::test] +async fn usage_one_token_over_the_bound_faults_and_stores_no_record() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let prepared = prepared_for("one over", 1_200); + let bound = bound_of(&prepared); + let inner = ScriptedPort::new(vec![Ok( + reply("edge", "r1").with_usage(usage(bound, 1, None)) + )]); + let port = recording(&inner, &store, "run-edge", bound + 200); + let fault = port.infer(prepared).await.expect_err("one over"); + assert_eq!(fault.class(), InferenceFaultClass::BudgetExhausted); + assert_eq!(port.spent(), bound + 1, "charged in full"); + assert!( + store + .records(&RunId::new("run-edge").unwrap()) + .unwrap() + .is_empty() + ); +} + +/// Every field of the provider request is in the replay key except the two +/// ids that name the call. +#[test] +fn the_replay_key_covers_what_the_request_asks_and_not_what_names_it() { + let base = prepared_for("key", 1_200).invocation.request.clone(); + let digest = |edit: &dyn Fn(&mut CodexProviderRequest)| { + let mut request = base.clone(); + edit(&mut request); + content_digest(&request).unwrap() + }; + let same = digest(&|_| {}); + assert_eq!(same, content_digest(&base).unwrap()); + let changes: [(&str, Box); 10] = [ + ("model", Box::new(|r| r.model.push('x'))), + ("instructions", Box::new(|r| r.instructions.push('x'))), + ( + "tools", + Box::new(|r| { + r.tools.push(CodexToolDefinition { + name: "t".into(), + description: "d".into(), + parameters_json: "{}".into(), + }) + }), + ), + ( + "input", + Box::new(|r| r.input.push(CodexInputItem::UserText { text: "x".into() })), + ), + ( + "max_output_tokens", + Box::new(|r| r.max_output_tokens = Some(1)), + ), + ( + "reasoning_effort", + Box::new(|r| r.reasoning_effort = Some("low".into())), + ), + ( + "service_tier", + Box::new(|r| r.service_tier = Some("flex".into())), + ), + ( + "output_schema_json", + Box::new(|r| r.output_schema_json = Some("{}".into())), + ), + ( + "previous_response_id", + Box::new(|r| r.previous_response_id = Some("p".into())), + ), + ( + "parallel_tool_calls", + Box::new(|r| r.parallel_tool_calls = !r.parallel_tool_calls), + ), + ]; + for (field, change) in changes { + assert_ne!(digest(&*change), same, "{field} is not in the replay key"); + } + assert_eq!(digest(&|r| r.request_id.push('x')), same); + assert_eq!(digest(&|r| r.conversation_id.push('x')), same); +} + +fn open_with( + rows: Vec, +) -> (tempfile::TempDir, Result) { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("calls.redb"); + let mut backing = OwnedRedbMessagePackBackingStore::new(&path).unwrap(); + for row in &rows { + backing.push(row).unwrap(); + } + drop(backing); + let opened = CallRecordStore::open(&path); + (directory, opened) +} + +fn record_row(record: &ProviderCallRecord) -> CultCacheEnvelope { + envelope( + CALL_RECORD_ROW, + Some(CALL_RECORD_SCHEMA), + &row_key(&record.run_id, record.sequence), + rmp_serde::to_vec_named(record).unwrap(), + ) +} + +#[test] +fn a_run_with_sequence_zero_beside_a_hole_is_refused() { + let (_directory, opened) = open_with(vec![ + record_row(&sample_record("run-x", 0)), + record_row(&sample_record("run-x", 2)), + spend_row("run-x", 6), + ]); + assert_eq!(opened.err(), Some(CallRecordError::Corrupt)); +} + +#[test] +fn a_record_charged_nothing_still_needs_its_ledger() { + let mut record = sample_record("run-y", 1); + record.charged = 0; + let (_directory, opened) = open_with(vec![record_row(&record)]); + assert_eq!(opened.err(), Some(CallRecordError::Corrupt)); +} + +#[test] +fn a_record_whose_run_id_is_no_whole_key_segment_is_refused() { + let mut stray = sample_record("run-z/b", 1); + stray.charged = 0; + let (_directory, opened) = open_with(vec![ + record_row(&stray), + record_row(&sample_record("run-z", 1)), + spend_row("run-z", 3), + ]); + assert_eq!(opened.err(), Some(CallRecordError::Corrupt)); + // A neighbouring run id that merely shares the prefix keeps its own records. + let (_directory, opened) = open_with(vec![ + record_row(&sample_record("run-z", 1)), + spend_row("run-z", 3), + spend_row("run-z-b", 0), + ]); + let store = opened.unwrap(); + assert_eq!( + store.records(&RunId::new("run-z").unwrap()).unwrap().len(), + 1 + ); +} + +#[tokio::test] +async fn a_run_reopens_only_under_the_cap_it_began_with() { + let directory = tempfile::tempdir().unwrap(); + let (store, path) = store_in(&directory); + let inner = ScriptedPort::new(Vec::new()); + drop(recording(&inner, &store, "run-cap", 5_000)); + let run = RunId::new("run-cap").unwrap(); + let open = |store: &Arc, cap: u64| { + RecordingInferencePort::open(inner.clone(), store.clone(), run.clone(), cap, &start()) + }; + for other in [4_999, 5_001, u64::MAX] { + assert_eq!( + open(&store, other).err(), + Some(CallRecordError::RunMismatch(RunField::Cap)) + ); + } + // A refused reopen does not hold the run, and the cap outlives the process. + drop(open(&store, 5_000).expect("the same cap reopens")); + drop(store); + let store = Arc::new(CallRecordStore::open(&path).unwrap()); + assert_eq!( + open(&store, 5_001).err(), + Some(CallRecordError::RunMismatch(RunField::Cap)) + ); + open(&store, 5_000).expect("the same cap reopens after a restart"); + let message = format!("{}", CallRecordError::RunMismatch(RunField::Cap)); + assert!(message.contains("token cap") && !message.contains("5000")); +} + +/// A world made by the production path: one genesis command in a fresh world +/// file, its snapshot, and the file's path. +async fn genesis( + directory: &std::path::Path, + command: CommandId, +) -> (crate::WorldSnapshot, std::path::PathBuf) { + let path = directory.join("world.cc"); + let (mailbox, task) = crate::WorldMailbox::open(&path).expect("an empty world"); + mailbox + .create_fixture( + crate::tests::creation(command, "Replayed"), + &crate::tests::auth_principal(crate::tests::owner()), + ) + .await + .expect("a created world"); + let snapshot = mailbox.snapshot().await.unwrap(); + drop(mailbox); + task.await.unwrap(); + (snapshot, path) +} + +async fn reopened(path: &std::path::Path) -> crate::WorldSnapshot { + let (mailbox, task) = crate::WorldMailbox::open(path).expect("a restored world"); + let snapshot = mailbox.snapshot().await.unwrap(); + drop(mailbox); + task.await.unwrap(); + snapshot +} + +#[tokio::test] +async fn replay_runs_against_a_restored_copy_and_refuses_any_other_world_up_front() { + let command = CommandId::new(); + let first = tempfile::tempdir().unwrap(); + let (world, world_file) = genesis(first.path(), command).await; + let began = RunStart::of(&world); + + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let run = RunId::new("run-world").unwrap(); + let inner = ScriptedPort::new(ok_all(&[reply("one", "r1")])); + let live = + RecordingInferencePort::open(inner.clone(), store.clone(), run.clone(), u64::MAX, &began) + .unwrap(); + live.infer(live.prepare(request_for(command, "go", 1_200)).unwrap()) + .await + .unwrap(); + drop(live); + + // A byte copy of the starting world, restored, replays the run. + let restored_dir = tempfile::tempdir().unwrap(); + let restored_file = restored_dir.path().join("world.cc"); + std::fs::copy(&world_file, &restored_file).unwrap(); + let restored = RunStart::of(&reopened(&restored_file).await); + assert_eq!(restored, began); + let replay = ReplayInferencePort::new("verse-replay", &store, &run, &restored) + .expect("a restored copy replays"); + let served = replay + .infer(replay.prepare(request_for(command, "go", 1_200)).unwrap()) + .await + .unwrap(); + assert_eq!(json(&served), json(&reply("one", "r1"))); + + // The same genesis created again is another world and is refused before + // any call, naming the field and nothing it held. + let again = tempfile::tempdir().unwrap(); + let (recreated, _) = genesis(again.path(), command).await; + let refusal = ReplayInferencePort::new("verse-replay", &store, &run, &RunStart::of(&recreated)) + .err() + .expect("a re-created world"); + assert_eq!(refusal, CallRecordError::RunMismatch(RunField::WorldId)); + let text = format!("{refusal} {refusal:?}"); + assert!(!text.contains(&began.world_id) && !text.contains(&began.state_digest)); + + // The same id with another digest is refused too. + let drifted = RunStart { + world_id: began.world_id.clone(), + state_digest: format!("{}x", began.state_digest), + }; + assert_eq!( + ReplayInferencePort::new("verse-replay", &store, &run, &drifted).err(), + Some(CallRecordError::RunMismatch(RunField::StateDigest)) + ); + assert_eq!(inner.calls(), 1, "replay never reached the provider"); +} diff --git a/crates/ghostlight/src/verse/mod.rs b/crates/ghostlight/src/verse/mod.rs index 7c7b6fb..ba0cd08 100644 --- a/crates/ghostlight/src/verse/mod.rs +++ b/crates/ghostlight/src/verse/mod.rs @@ -6,5 +6,5 @@ mod calls; pub use calls::{ CALL_RECORD_SCHEMA, CallRecordError, CallRecordStore, ProviderCallRecord, - RecordingInferencePort, ReplayInferencePort, RunId, + RecordingInferencePort, ReplayInferencePort, RunField, RunId, RunStart, }; From e651d9e863ca7ad2467ec58e8eafbc0237c9f5b3 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 10 Oct 2026 20:33:27 +0200 Subject: [PATCH 08/10] Call record: fix the moved ledger borrow (previous commit does not build) Co-Authored-By: Claude Opus 5.5 --- crates/ghostlight/src/verse/calls.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/crates/ghostlight/src/verse/calls.rs b/crates/ghostlight/src/verse/calls.rs index 4392114..d86a92d 100644 --- a/crates/ghostlight/src/verse/calls.rs +++ b/crates/ghostlight/src/verse/calls.rs @@ -339,7 +339,8 @@ impl CallRecordStore { tally.last_sequence = tally.last_sequence.max(record.sequence); } DecodedRow::Ledger(ledger) => { - runs.entry(ledger.run_id.clone()).or_default().ledger = Some(ledger); + let run = ledger.run_id.clone(); + runs.entry(run).or_default().ledger = Some(ledger); } } } @@ -446,7 +447,8 @@ impl CallRecordStore { schema_id: Some(RUN_LEDGER_SCHEMA.into()), }; state.store.push(&row).map_err(|_| CallRecordError::Store)?; - state.runs.entry(ledger.run_id.clone()).or_default().ledger = Some(ledger); + let run = ledger.run_id.clone(); + state.runs.entry(run).or_default().ledger = Some(ledger); Ok(()) } From ecff1ac25bfe3c6bc41a3bee7785e19454c1d331 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 10 Oct 2026 20:46:21 +0200 Subject: [PATCH 09/10] Call record tests: resumed and exact-cap cases keep one cap per run Co-Authored-By: Claude Opus 5.5 --- crates/ghostlight/src/verse/calls/tests.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/ghostlight/src/verse/calls/tests.rs b/crates/ghostlight/src/verse/calls/tests.rs index ca88565..30a51e8 100644 --- a/crates/ghostlight/src/verse/calls/tests.rs +++ b/crates/ghostlight/src/verse/calls/tests.rs @@ -270,7 +270,7 @@ async fn cap_refuses_before_the_provider_is_called() { drop(port); // Exactly the cap is allowed. - let exact = recording(&inner, &store, "run-cap", bound); + let exact = recording(&inner, &store, "run-exact-cap", bound); exact .infer(probe) .await @@ -592,13 +592,13 @@ async fn a_reopened_run_resumes_its_spend() { let inner = ScriptedPort::new(vec![Ok( reply("one", "r1").with_usage(usage(bound, 0, None)) )]); - let first = recording(&inner, &store, "run-resume", bound * 2); + let first = recording(&inner, &store, "run-resume", bound * 2 - 1); first.infer(probe.clone()).await.unwrap(); drop((first, store)); let store = Arc::new(CallRecordStore::open(&path).unwrap()); let inner = ScriptedPort::new(vec![Ok(reply("two", "r2"))]); - let resumed = recording(&inner, &store, "run-resume", bound * 2); + let resumed = recording(&inner, &store, "run-resume", bound * 2 - 1); assert_eq!(resumed.spent(), bound); let fault = resumed .infer(probe) From 308169ac51c03969bb7157f0d2c1eb4983a77f69 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 10 Oct 2026 22:31:20 +0200 Subject: [PATCH 10/10] call-record: pin all fifteen keyed fields of the replay key The replay-key test now covers schema_id, reasoning_summary, output_format_name, tool_choice and prompt_cache_key beside the ten it had, over a Vec so a future keyed field is one more row. Tests only; every commit builds. Co-Authored-By: Claude Opus 5.5 --- crates/ghostlight/src/verse/calls/tests.rs | 28 +++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/crates/ghostlight/src/verse/calls/tests.rs b/crates/ghostlight/src/verse/calls/tests.rs index 30a51e8..02131ae 100644 --- a/crates/ghostlight/src/verse/calls/tests.rs +++ b/crates/ghostlight/src/verse/calls/tests.rs @@ -976,7 +976,7 @@ fn the_replay_key_covers_what_the_request_asks_and_not_what_names_it() { }; let same = digest(&|_| {}); assert_eq!(same, content_digest(&base).unwrap()); - let changes: [(&str, Box); 10] = [ + let changes: Vec<(&str, Box)> = vec![ ("model", Box::new(|r| r.model.push('x'))), ("instructions", Box::new(|r| r.instructions.push('x'))), ( @@ -1017,6 +1017,32 @@ fn the_replay_key_covers_what_the_request_asks_and_not_what_names_it() { "parallel_tool_calls", Box::new(|r| r.parallel_tool_calls = !r.parallel_tool_calls), ), + ("schema_id", Box::new(|r| r.schema_id.push_str(".v2"))), + ( + "reasoning_summary", + Box::new(|r| r.reasoning_summary = Some("auto".into())), + ), + ( + "output_format_name", + Box::new(|r| r.output_format_name = Some("persona_reply".into())), + ), + ( + "tool_choice", + Box::new(|r| { + r.tool_choice = match r.tool_choice { + codex_connector::CodexToolChoice::Auto => { + codex_connector::CodexToolChoice::Required + } + codex_connector::CodexToolChoice::Required => { + codex_connector::CodexToolChoice::Auto + } + } + }), + ), + ( + "prompt_cache_key", + Box::new(|r| r.prompt_cache_key = Some("ghostlight-persona-1".into())), + ), ]; for (field, change) in changes { assert_ne!(digest(&*change), same, "{field} is not in the replay key");