diff --git a/crates/ghostlight/src/controllers.rs b/crates/ghostlight/src/controllers.rs index 41bf671..990de62 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 5f124a8..c28dbb3 100644 --- a/crates/ghostlight/src/lib.rs +++ b/crates/ghostlight/src/lib.rs @@ -71,6 +71,7 @@ mod sdk_inference; mod table; mod tool_schema; mod vault; +mod verse; pub(crate) use action::ActionMismatch; pub(crate) use clock::Motion; @@ -135,6 +136,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, RunField, RunId, RunStart, +}; use patch::{EdgeRecord, EntityRecord, LedgerDelta, ResolvedOp, ResolvedPatch}; use chrono::{DateTime, Utc}; 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); diff --git a/crates/ghostlight/src/verse/calls.rs b/crates/ghostlight/src/verse/calls.rs new file mode 100644 index 0000000..d86a92d --- /dev/null +++ b/crates/ghostlight/src/verse/calls.rs @@ -0,0 +1,749 @@ +//! 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 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, 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, BTreeSet, VecDeque}, + path::Path, + sync::{Arc, Mutex}, +}; +use thiserror::Error; + +const CALL_RECORD_ROW: &str = "verse_provider_call.v1"; +const RUN_LEDGER_ROW: &str = "verse_run_ledger.v1"; +/// The run ledger document's schema. +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. +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, + #[error("another recording port already owns this run")] + 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. +#[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 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)] +pub struct ProviderCallRecord { + run_id: String, + sequence: u64, + purpose: InferencePurpose, + model: String, + content_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 request's content digest (see `content_digest`). + pub fn content_sha256(&self) -> &str { + &self.content_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]) -> String { + digest.iter().map(|byte| format!("{byte:02x}")).collect() +} + +/// 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 { + // 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(&( + 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, 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 RunLedger { + run_id: String, + cap: u64, + start: RunStart, + spent: u64, +} + +enum DecodedRow { + Call(ProviderCallRecord), + Ledger(RunLedger), +} + +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) + } +} + +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)?; + // 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_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::Ledger(ledger)) + } + _ => Err(CallRecordError::Corrupt), + } +} + +#[derive(Default)] +struct RunTally { + last_sequence: u64, + /// The persisted ledger, read at open and written where it is spent. + ledger: Option, +} + +struct StoreState { + store: OwnedRedbMessagePackBackingStore, + runs: BTreeMap, + /// Runs with a recording port alive; one ledger has one owner. + owned: BTreeSet, +} + +/// 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 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)?; + store + .validate_path_identity() + .map_err(|_| CallRecordError::Store)?; + 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 { + 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::Ledger(ledger) => { + let run = ledger.run_id.clone(); + runs.entry(run).or_default().ledger = Some(ledger); + } + } + } + // 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); + let covered = tally + .ledger + .as_ref() + .is_some_and(|ledger| ledger.spent >= recorded); + if rows != tally.last_sequence || !covered { + return Err(CallRecordError::Corrupt); + } + } + Ok(Self { + state: Mutex::new(StoreState { + store, + runs, + owned: BTreeSet::new(), + }), + }) + } + + 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 has reserved or been charged: its persisted ledger. + pub fn spent(&self, run: &RunId) -> Result { + Ok(self + .lock()? + .runs + .get(run.as_str()) + .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. + /// 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) { + if let Ok(mut state) = self.state.lock() { + state.owned.remove(run.as_str()); + } + } + + /// 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: 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_LEDGER_SCHEMA.into()), + }; + state.store.push(&row).map_err(|_| CallRecordError::Store)?; + let run = ledger.run_id.clone(); + state.runs.entry(run).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 after = Self::held(&state, run).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 settled = Self::held(&state, run) + .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) + .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)?; + state + .runs + .entry(record.run_id.clone()) + .or_default() + .last_sequence = record.sequence; + 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)) { + if let DecodedRow::Call(record) = decode_row(row)? { + records.push(record); + } + } + records.sort_by_key(|record| record.sequence); + Ok(records) + } +} + +/// 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, +} + +impl RecordingInferencePort { + /// 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.claim(&run, cap_tokens, start)?; + Ok(Self { + inner, + store, + run, + cap: cap_tokens, + }) + } + + /// Tokens reserved or charged to the run so far. + pub fn spent(&self) -> u64 { + self.store.spent(&self.run).unwrap_or(u64::MAX) + } + + /// 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 settle( + &self, + bound: u64, + identity: (InferencePurpose, String, String), + result: Result, + ) -> Result { + // 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, + }; + 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: self.store.next_sequence(&self.run).map_err(unrecorded)?, + purpose, + model, + content_sha256, + events: output.events.clone(), + receipt_digest: output.receipt_digest.clone(), + usage: output.usage, + charged, + }; + 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") +} + +#[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)?; + let identity = ( + request.purpose, + request.invocation.request.model.clone(), + 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) + } + + 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, 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 { + caller_runtime_id: String, + pending: Mutex>, +} + +impl ReplayInferencePort { + /// 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, + store: &CallRecordStore, + run: &RunId, + world: &RunStart, + ) -> Result { + 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(store.records(run)?.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 = content_digest(&request.invocation.request)?; + let Some(next) = pending.front() else { + return Err(InferenceFault::integrity_violation( + "replay has no recorded call left", + )); + }; + if next.content_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..02131ae --- /dev/null +++ b/crates/ghostlight/src/verse/calls/tests.rs @@ -0,0 +1,1230 @@ +use super::*; +use crate::CommandId; +use crate::controllers::{ + InferenceFaultClass, 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, + lent: AtomicUsize, + gate: Option>, +} + +impl ScriptedPort { + fn new(replies: Script) -> Arc { + Arc::new(Self { + replies: Mutex::new(replies.into()), + calls: AtomicUsize::new(0), + lent: AtomicUsize::new(0), + gate: None, + }) + } + + fn gated(replies: Script, gate: Arc) -> Arc { + Arc::new(Self { + replies: Mutex::new(replies.into()), + calls: AtomicUsize::new(0), + lent: 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) + } + + 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 { + 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, + ) +} + +/// 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, + &start(), + ) + .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 = 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(); + 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.content_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); + drop(port); + + // Exactly the cap is allowed. + let exact = recording(&inner, &store, "run-exact-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 }) + }; + 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), + 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(); + 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_keeps_its_reservation_charged() { + 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)), + Err(InferenceFault::new("junk").classed(InferenceFaultClass::BadReply)), + ]); + let port = recording(&inner, &store, "run-fault", u64::MAX); + for (index, class) in [ + InferenceFaultClass::Connect, + InferenceFaultClass::Timeout, + 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()) + .unwrap() + .is_empty() + ); +} + +#[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, + &start(), + ); + 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(); + 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 - 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 - 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 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)) + .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(), + content_sha256: "ab".into(), + events: vec![InferenceEvent::Text("t".into())], + receipt_digest: "r".into(), + usage: None, + charged: 3, + } +} + +#[test] +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) + ); +} + +#[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(), 2, "one record and the run's ledger"); + 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 spend_row(run: &str, spent: u64) -> CultCacheEnvelope { + envelope( + RUN_LEDGER_ROW, + Some(RUN_LEDGER_SCHEMA), + run, + rmp_serde::to_vec_named(&RunLedger { + run_id: run.into(), + cap: u64::MAX, + start: start(), + spent, + }) + .unwrap(), + ) +} + +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])), + ( + "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)), + ("a record with no ledger", record_at(1)), + ( + "a ledger at the wrong key", + envelope( + RUN_LEDGER_ROW, + Some(RUN_LEDGER_SCHEMA), + "other", + rmp_serde::to_vec_named(&RunLedger { + run_id: "run-x".into(), + cap: u64::MAX, + start: start(), + spent: 1, + }) + .unwrap(), + ), + ), + ( + "a ledger that is not canonical", + envelope( + RUN_LEDGER_ROW, + Some(RUN_LEDGER_SCHEMA), + "run-x", + [ + rmp_serde::to_vec_named(&RunLedger { + run_id: "run-x".into(), + cap: u64::MAX, + start: start(), + spent: 1, + }) + .unwrap(), + vec![0xc0], + ] + .concat(), + ), + ), + ]; + 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}"); + } + // A ledger under the charges of its records is refused too. + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("calls.redb"); + 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.spent(&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)); + } +} + +#[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, &run, &start()).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"))); +} + +#[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: Vec<(&str, Box)> = vec![ + ("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), + ), + ("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"); + } + 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 new file mode 100644 index 0000000..ba0cd08 --- /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, RunField, RunId, RunStart, +};