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, +};