GameCult/Ghostlight
Watch
9
0
Fork
You've already forked Ghostlight
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Verse call records: a recording port owns the run's spend, a replay port serves the stored responses

Browse source 
ProviderCallRecord (ghostlight.verse_provider_call.v1) keyed (run id, sequence),
CallRecordStore over an owned redb CultCache file, RecordingInferencePort
(bound check before the call, real usage charged after, record written before
the output returns) and ReplayInferencePort. InferenceFaultClass gains
BudgetExhausted. Not yet compiled or run: this commit may not build.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-10 14:02:48 +02:00
commit e28bf3fd55
5 changed files with 1152 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

11
crates/ghostlight/src/controllers.rs
View file

@ -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;

5
crates/ghostlight/src/lib.rs
View file

@ -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};

474
crates/ghostlight/src/verse/calls.rs Normal file
View file

@ -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<Self, CallRecordError> {
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<InferenceEvent>,
receipt_digest: String,
usage: Option<TokenUsage>,
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<TokenUsage> {
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<ProviderCallRecord, CallRecordError> {
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<String, RunTally>,
}
/// 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<StoreState>,
}
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<Path>) -> Result<Self, CallRecordError> {
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<String, RunTally> = 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<std::sync::MutexGuard<'_, StoreState>, 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<u64, CallRecordError> {
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<u64, CallRecordError> {
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<Vec<ProviderCallRecord>, 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<dyn InferencePort>,
store: Arc<CallRecordStore>,
run: RunId,
cap: u64,
ledger: Mutex<Ledger>,
}
impl RecordingInferencePort {
/// The run's spend resumes from its stored records.
pub fn open(
inner: Arc<dyn InferencePort>,
store: Arc<CallRecordStore>,
run: RunId,
cap_tokens: u64,
) -> Result<Self, CallRecordError> {
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<u64, InferenceFault> {
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<InferenceOutput, InferenceFault>,
) -> Result<InferenceOutput, InferenceFault> {
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<PreparedInference, InferenceFault> {
self.inner.prepare(request)
}
async fn infer(&self, request: PreparedInference) -> Result<InferenceOutput, InferenceFault> {
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<dyn ToolResultOracle>) {
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<VecDeque<ProviderCallRecord>>,
}
impl ReplayInferencePort {
/// `records` are one run's, as `CallRecordStore::records` returns them.
pub fn new(
caller_runtime_id: impl Into<String>,
records: Vec<ProviderCallRecord>,
) -> Result<Self, CallRecordError> {
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, InferenceFault> {
PreparedInference::prepare(
&self.caller_runtime_id,
unix_ms()?.saturating_add(REQUEST_EXPIRY.as_millis() as u64),
request,
)
}
async fn infer(&self, request: PreparedInference) -> Result<InferenceOutput, InferenceFault> {
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;

652
crates/ghostlight/src/verse/calls/tests.rs Normal file
View file

@ -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<Result<InferenceOutput, InferenceFault>>;
/// 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<VecDeque<Result<InferenceOutput, InferenceFault>>>,
calls: AtomicUsize,
gate: Option<Arc<tokio::sync::Semaphore>>,
}
impl ScriptedPort {
fn new(replies: Script) -> Arc<Self> {
Arc::new(Self {
replies: Mutex::new(replies.into()),
calls: AtomicUsize::new(0),
gate: None,
})
}
fn gated(replies: Script, gate: Arc<tokio::sync::Semaphore>) -> Arc<Self> {
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, InferenceFault> {
PreparedInference::prepare("verse-test", 4_102_444_800_000, request)
}
async fn infer(&self, _: PreparedInference) -> Result<InferenceOutput, InferenceFault> {
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::<CodexToolDefinition>::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<u64>) -> TokenUsage {
TokenUsage {
prompt,
completion,
cached_prompt,
}
}
fn store_in(directory: &tempfile::TempDir) -> (Arc<CallRecordStore>, std::path::PathBuf) {
let path = directory.path().join("calls.redb");
(
Arc::new(CallRecordStore::open(&path).expect("the store opens")),
path,
)
}
fn recording(
inner: &Arc<ScriptedPort>,
store: &Arc<CallRecordStore>,
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<u8> {
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<u64> = 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::<u64>());
// 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<u8>,
) -> 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<u8>| {
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));
}
}

10
crates/ghostlight/src/verse/mod.rs Normal file
View file

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