From 2c5eb95da3f1b0d1d9a2fdc490fe04fa96205b7e Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 10 Oct 2026 20:32:17 +0200 Subject: [PATCH] Call record: ledger carries cap and starting world; replay refuses another world The run ledger is the one owner of what a run began as: its cap and the world id and state digest it started from. Replay takes the store, the run and a restored world and refuses a different world up front with a typed error naming the field. Restores the decode checks the ledger rules do not imply, makes the replay key an exhaustive destructure, and pins the over-bound edge. Not yet built or tested at this commit. Co-Authored-By: Claude Opus 5.5 --- crates/ghostlight/src/lib.rs | 2 +- crates/ghostlight/src/verse/calls.rs | 264 ++++++++++++----- crates/ghostlight/src/verse/calls/tests.rs | 330 +++++++++++++++++++-- crates/ghostlight/src/verse/mod.rs | 2 +- 4 files changed, 499 insertions(+), 99 deletions(-) diff --git a/crates/ghostlight/src/lib.rs b/crates/ghostlight/src/lib.rs index 409a251..ce1ba8f 100644 --- a/crates/ghostlight/src/lib.rs +++ b/crates/ghostlight/src/lib.rs @@ -133,7 +133,7 @@ pub use table::{ pub use vault::{VaultError, VaultEvidenceSource}; pub use verse::{ CALL_RECORD_SCHEMA, CallRecordError, CallRecordStore, ProviderCallRecord, RecordingInferencePort, - ReplayInferencePort, RunId, + ReplayInferencePort, RunField, RunId, RunStart, }; use patch::{EdgeRecord, EntityRecord, LedgerDelta, ResolvedOp, ResolvedPatch}; diff --git a/crates/ghostlight/src/verse/calls.rs b/crates/ghostlight/src/verse/calls.rs index dbf7e9d..4392114 100644 --- a/crates/ghostlight/src/verse/calls.rs +++ b/crates/ghostlight/src/verse/calls.rs @@ -38,9 +38,9 @@ use std::{ use thiserror::Error; const CALL_RECORD_ROW: &str = "verse_provider_call.v1"; -const RUN_SPEND_ROW: &str = "verse_run_spend.v1"; +const RUN_LEDGER_ROW: &str = "verse_run_ledger.v1"; /// The run ledger document's schema. -const RUN_SPEND_SCHEMA: &str = "ghostlight.verse_run_spend.v1"; +const RUN_LEDGER_SCHEMA: &str = "ghostlight.verse_run_ledger.v1"; /// The call record document's schema. A store holding any other row type or /// schema is refused at open: this store is not migrated from, or shared with, /// another document. @@ -62,6 +62,46 @@ pub enum CallRecordError { RunInUse, #[error("the call could pass the run's token cap")] OverCap, + #[error("the run was recorded with a different {0}")] + RunMismatch(RunField), +} + +/// What a run is bound to when it begins, named by the field a mismatch refuses. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum RunField { + /// The token cap the run was opened with. + Cap, + /// The identity of the world the run started from. + WorldId, + /// The state digest of the world the run started from. + StateDigest, +} + +impl std::fmt::Display for RunField { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + Self::Cap => "token cap", + Self::WorldId => "world id", + Self::StateDigest => "world state digest", + }) + } +} + +/// The world a run began from: its identity and state digest. Replay is of a +/// restored copy of this world, so it takes the same two values. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RunStart { + world_id: String, + state_digest: String, +} + +impl RunStart { + pub fn of(world: &crate::WorldSnapshot) -> Self { + Self { + world_id: world.world_id.text(), + state_digest: world.state_digest.clone(), + } + } } /// A run's identity as a key segment. @@ -158,37 +198,62 @@ fn hex(digest: &[u8]) -> String { /// one world's random identity. Everything else the provider is sent /// (model, instructions, input, tools, shape) is in the digest. fn content_digest(request: &CodexProviderRequest) -> Result { + // Exhaustive on purpose: a field added upstream must be placed in the key + // or left out here by name before this compiles. + let CodexProviderRequest { + schema_id, + request_id: _request_id, // names the call, not what it asks + conversation_id: _conversation_id, // names the call, not what it asks + model, + instructions, + input, + reasoning_effort, + reasoning_summary, + service_tier, + output_format_name, + previous_response_id, + tools, + tool_choice, + parallel_tool_calls, + output_schema_json, + max_output_tokens, + prompt_cache_key, + } = request; let bytes = rmp_serde::to_vec(&( - &request.schema_id, - &request.model, - &request.instructions, - &request.input, - &request.reasoning_effort, - &request.reasoning_summary, - &request.service_tier, - &request.output_format_name, - &request.previous_response_id, - &request.tools, - &request.tool_choice, - request.parallel_tool_calls, - &request.output_schema_json, - request.max_output_tokens, - &request.prompt_cache_key, + schema_id, + model, + instructions, + input, + reasoning_effort, + reasoning_summary, + service_tier, + output_format_name, + previous_response_id, + tools, + tool_choice, + parallel_tool_calls, + output_schema_json, + max_output_tokens, + prompt_cache_key, )) .map_err(|_| InferenceFault::new("a provider request could not be measured"))?; Ok(hex(&Sha256::digest(bytes))) } -/// A run's persisted ledger: tokens reserved or charged against its cap. +/// A run's persisted ledger, the one owner of what the run began as: the cap +/// it was opened under, the world it started from, and the tokens reserved or +/// charged so far. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -struct RunSpend { +struct RunLedger { run_id: String, + cap: u64, + start: RunStart, spent: u64, } enum DecodedRow { Call(ProviderCallRecord), - Spend(RunSpend), + Ledger(RunLedger), } fn canonical Deserialize<'de>>( @@ -207,20 +272,22 @@ fn decode_row(row: &CultCacheEnvelope) -> Result { match (row.r#type.as_str(), row.schema_id.as_deref()) { (CALL_RECORD_ROW, Some(CALL_RECORD_SCHEMA)) => { let record: ProviderCallRecord = canonical(row)?; - // The run id and a sequence of zero need no check here: a run id - // that is no key segment has no readable ledger, and a zero - // sequence fails the density rule in `open`. - if row.key != row_key(&record.run_id, record.sequence) { + // A run id that is no whole key segment would key into another + // run's records, and sequences count from 1. + if RunId::new(&record.run_id).is_err() + || record.sequence == 0 + || row.key != row_key(&record.run_id, record.sequence) + { return Err(CallRecordError::Corrupt); } Ok(DecodedRow::Call(record)) } - (RUN_SPEND_ROW, Some(RUN_SPEND_SCHEMA)) => { - let spend: RunSpend = canonical(row)?; - if RunId::new(&spend.run_id).is_err() || row.key != spend.run_id { + (RUN_LEDGER_ROW, Some(RUN_LEDGER_SCHEMA)) => { + let ledger: RunLedger = canonical(row)?; + if RunId::new(&ledger.run_id).is_err() || row.key != ledger.run_id { return Err(CallRecordError::Corrupt); } - Ok(DecodedRow::Spend(spend)) + Ok(DecodedRow::Ledger(ledger)) } _ => Err(CallRecordError::Corrupt), } @@ -229,8 +296,8 @@ fn decode_row(row: &CultCacheEnvelope) -> Result { #[derive(Default)] struct RunTally { last_sequence: u64, - /// Tokens reserved or charged: the persisted ledger, read at open. - spent: u64, + /// The persisted ledger, read at open and written where it is spent. + ledger: Option, } struct StoreState { @@ -271,18 +338,23 @@ impl CallRecordStore { let tally = runs.entry(record.run_id).or_default(); tally.last_sequence = tally.last_sequence.max(record.sequence); } - DecodedRow::Spend(spend) => { - runs.entry(spend.run_id).or_default().spent = spend.spent; + DecodedRow::Ledger(ledger) => { + runs.entry(ledger.run_id.clone()).or_default().ledger = Some(ledger); } } } - // Keys are unique, so a run is dense from 1 exactly when its rows - // number its highest sequence. Every record was reserved before it was - // made, so the ledger covers what the records charged. + // Sequences start at 1 and keys are unique, so a run is dense exactly + // when its rows number its highest sequence. Every record was reserved + // in a ledger before it was made, so a run with records has a ledger + // that covers what they charged. for (run, tally) in &runs { let rows = counts.get(run).copied().unwrap_or(0); let recorded = charged.get(run).copied().unwrap_or(0); - if rows != tally.last_sequence || tally.spent < recorded { + let covered = tally + .ledger + .as_ref() + .is_some_and(|ledger| ledger.spent >= recorded); + if rows != tally.last_sequence || !covered { return Err(CallRecordError::Corrupt); } } @@ -315,16 +387,46 @@ impl CallRecordStore { .lock()? .runs .get(run.as_str()) - .map_or(0, |tally| tally.spent)) + .and_then(|tally| tally.ledger.as_ref()) + .map_or(0, |ledger| ledger.spent)) } /// Takes the run's ledger for one port, or refuses when another holds it. - fn own(&self, run: &RunId) -> Result<(), CallRecordError> { - if self.lock()?.owned.insert(run.as_str().to_owned()) { - Ok(()) - } else { - Err(CallRecordError::RunInUse) + /// The first claim of a run writes its ledger: the cap it is opened under + /// and the world it starts from. A later claim must name the same cap; the + /// start is not compared, because a resumed world has moved on. + fn claim(&self, run: &RunId, cap: u64, start: &RunStart) -> Result<(), CallRecordError> { + let mut state = self.lock()?; + if !state.owned.insert(run.as_str().to_owned()) { + return Err(CallRecordError::RunInUse); } + let begun = match state.runs.get(run.as_str()).and_then(|t| t.ledger.as_ref()) { + Some(ledger) if ledger.cap == cap => Ok(()), + Some(_) => Err(CallRecordError::RunMismatch(RunField::Cap)), + None => Self::write_ledger( + &mut state, + RunLedger { + run_id: run.as_str().to_owned(), + cap, + start: start.clone(), + spent: 0, + }, + ), + }; + if begun.is_err() { + state.owned.remove(run.as_str()); + } + begun + } + + /// The world a run began from, when it has begun. + fn start_of(&self, run: &RunId) -> Result, CallRecordError> { + Ok(self + .lock()? + .runs + .get(run.as_str()) + .and_then(|tally| tally.ledger.as_ref()) + .map(|ledger| ledger.start.clone())) } fn disown(&self, run: &RunId) { @@ -333,32 +435,46 @@ impl CallRecordStore { } } - /// Persists `spent` as the run's ledger, then adopts it in memory. A write - /// that fails leaves the previous figure. - fn write_spent(state: &mut StoreState, run: &str, spent: u64) -> Result<(), CallRecordError> { - let spend = RunSpend { - run_id: run.to_owned(), - spent, - }; + /// Persists the ledger, then adopts it in memory. A write that fails + /// leaves the previous ledger. + fn write_ledger(state: &mut StoreState, ledger: RunLedger) -> Result<(), CallRecordError> { let row = CultCacheEnvelope { - key: run.to_owned(), - r#type: RUN_SPEND_ROW.into(), - payload: rmp_serde::to_vec_named(&spend).map_err(|_| CallRecordError::Store)?, + key: ledger.run_id.clone(), + r#type: RUN_LEDGER_ROW.into(), + payload: rmp_serde::to_vec_named(&ledger).map_err(|_| CallRecordError::Store)?, stored_at: Utc::now().to_rfc3339(), - schema_id: Some(RUN_SPEND_SCHEMA.into()), + schema_id: Some(RUN_LEDGER_SCHEMA.into()), }; state.store.push(&row).map_err(|_| CallRecordError::Store)?; - state.runs.entry(run.to_owned()).or_default().spent = spent; + state.runs.entry(ledger.run_id.clone()).or_default().ledger = Some(ledger); Ok(()) } + /// Persists `spent` as the run's spend; the run must have been claimed. + fn write_spent(state: &mut StoreState, run: &str, spent: u64) -> Result<(), CallRecordError> { + let mut ledger = state + .runs + .get(run) + .and_then(|tally| tally.ledger.clone()) + .ok_or(CallRecordError::Store)?; + ledger.spent = spent; + Self::write_ledger(state, ledger) + } + + fn held(state: &StoreState, run: &RunId) -> u64 { + state + .runs + .get(run.as_str()) + .and_then(|tally| tally.ledger.as_ref()) + .map_or(0, |ledger| ledger.spent) + } + /// Charges `bound` to the run's persisted ledger, or refuses with /// `OverCap` when the ledger plus the bound would pass `cap`. The check and /// the write hold one lock, so calls on the wire hold their bounds. fn reserve(&self, run: &RunId, bound: u64, cap: u64) -> Result<(), CallRecordError> { let mut state = self.lock()?; - let held = state.runs.get(run.as_str()).map_or(0, |tally| tally.spent); - let after = held.saturating_add(bound); + let after = Self::held(&state, run).saturating_add(bound); if after > cap { return Err(CallRecordError::OverCap); } @@ -379,8 +495,9 @@ impl CallRecordStore { if let Some(record) = record { Self::push_record(&mut state, record)?; } - let held = state.runs.get(run.as_str()).map_or(0, |tally| tally.spent); - let settled = held.saturating_sub(bound).saturating_add(charged); + let settled = Self::held(&state, run) + .saturating_sub(bound) + .saturating_add(charged); Self::write_spent(&mut state, run.as_str(), settled) } @@ -446,15 +563,18 @@ pub struct RecordingInferencePort { } impl RecordingInferencePort { - /// Takes ownership of the run's ledger, which resumes from the store. A - /// second port on a run whose first is alive is refused with `RunInUse`. + /// Takes ownership of the run's ledger, which resumes from the store. The + /// first open of a run records its cap and the world it starts from; a + /// reopen with another cap is refused with `RunMismatch(Cap)`. A second + /// port on a run whose first is alive is refused with `RunInUse`. pub fn open( inner: Arc, store: Arc, run: RunId, cap_tokens: u64, + start: &RunStart, ) -> Result { - store.own(&run)?; + store.claim(&run, cap_tokens, start)?; Ok(Self { inner, store, @@ -567,20 +687,26 @@ pub struct ReplayInferencePort { } impl ReplayInferencePort { - /// `records` are one run's, as `CallRecordStore::records` returns them. + /// Replays `run` from the store against `world`, a restored copy of the + /// world the run started from. Any other world is refused here, before a + /// call, with `RunMismatch` naming the field; a run the store never began + /// is `NotReplayable`. pub fn new( caller_runtime_id: impl Into, - records: Vec, + store: &CallRecordStore, + run: &RunId, + world: &RunStart, ) -> Result { - let dense = records.iter().enumerate().all(|(index, record)| { - record.sequence == index as u64 + 1 && record.run_id == records[0].run_id - }); - if !dense { - return Err(CallRecordError::NotReplayable); + let began = store.start_of(run)?.ok_or(CallRecordError::NotReplayable)?; + if began.world_id != world.world_id { + return Err(CallRecordError::RunMismatch(RunField::WorldId)); + } + if began.state_digest != world.state_digest { + return Err(CallRecordError::RunMismatch(RunField::StateDigest)); } Ok(Self { caller_runtime_id: caller_runtime_id.into(), - pending: Mutex::new(records.into()), + pending: Mutex::new(store.records(run)?.into()), }) } } diff --git a/crates/ghostlight/src/verse/calls/tests.rs b/crates/ghostlight/src/verse/calls/tests.rs index 8de27c4..ca88565 100644 --- a/crates/ghostlight/src/verse/calls/tests.rs +++ b/crates/ghostlight/src/verse/calls/tests.rs @@ -129,14 +129,28 @@ fn store_in(directory: &tempfile::TempDir) -> (Arc, std::path:: ) } +/// A world identity and digest for tests that need a run to start somewhere. +fn start() -> RunStart { + RunStart { + world_id: "world-a".into(), + state_digest: "sha256:a".into(), + } +} + fn recording( inner: &Arc, store: &Arc, run: &str, cap: u64, ) -> RecordingInferencePort { - RecordingInferencePort::open(inner.clone(), store.clone(), RunId::new(run).unwrap(), cap) - .expect("the port opens") + RecordingInferencePort::open( + inner.clone(), + store.clone(), + RunId::new(run).unwrap(), + cap, + &start(), + ) + .expect("the port opens") } fn json(output: &InferenceOutput) -> Vec { @@ -502,6 +516,7 @@ async fn one_run_has_one_ledger_and_one_owner() { store.clone(), RunId::new("run-own").unwrap(), cap, + &start(), ); assert_eq!(second.err(), Some(CallRecordError::RunInUse)); // Another run is another ledger. @@ -583,7 +598,7 @@ async fn a_reopened_run_resumes_its_spend() { let store = Arc::new(CallRecordStore::open(&path).unwrap()); let inner = ScriptedPort::new(vec![Ok(reply("two", "r2"))]); - let resumed = recording(&inner, &store, "run-resume", bound * 2 - 1); + let resumed = recording(&inner, &store, "run-resume", bound * 2); assert_eq!(resumed.spent(), bound); let fault = resumed .infer(probe) @@ -619,8 +634,7 @@ async fn replay_serves_stored_responses_and_refuses_strangers() { } let calls_made = inner.calls(); - let stored = store.records(&run).unwrap(); - let replay = ReplayInferencePort::new("verse-replay", stored).unwrap(); + let replay = ReplayInferencePort::new("verse-replay", &store, &run, &start()).unwrap(); // A request the run never made is a fault and consumes nothing. let stranger = replay .prepare(request_for(CommandId::new(), "z", 1_200)) @@ -665,20 +679,14 @@ fn sample_record(run: &str, sequence: u64) -> ProviderCallRecord { } #[test] -fn replay_needs_one_runs_dense_records() { - assert!( - ReplayInferencePort::new("x", vec![sample_record("a", 1), sample_record("a", 2)]).is_ok() +fn replay_needs_a_run_the_store_began() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let run = RunId::new("never-begun").unwrap(); + assert_eq!( + ReplayInferencePort::new("x", &store, &run, &start()).err(), + Some(CallRecordError::NotReplayable) ); - for records in [ - vec![sample_record("a", 2)], - vec![sample_record("a", 1), sample_record("a", 3)], - vec![sample_record("a", 1), sample_record("b", 2)], - ] { - assert_eq!( - ReplayInferencePort::new("x", records).err(), - Some(CallRecordError::NotReplayable) - ); - } } #[tokio::test] @@ -710,11 +718,13 @@ async fn a_record_holds_no_request_text() { fn spend_row(run: &str, spent: u64) -> CultCacheEnvelope { envelope( - RUN_SPEND_ROW, - Some(RUN_SPEND_SCHEMA), + RUN_LEDGER_ROW, + Some(RUN_LEDGER_SCHEMA), run, - rmp_serde::to_vec_named(&RunSpend { + rmp_serde::to_vec_named(&RunLedger { run_id: run.into(), + cap: u64::MAX, + start: start(), spent, }) .unwrap(), @@ -800,11 +810,13 @@ fn the_store_refuses_any_row_that_is_not_a_canonical_v1_record() { ( "a ledger at the wrong key", envelope( - RUN_SPEND_ROW, - Some(RUN_SPEND_SCHEMA), + RUN_LEDGER_ROW, + Some(RUN_LEDGER_SCHEMA), "other", - rmp_serde::to_vec_named(&RunSpend { + rmp_serde::to_vec_named(&RunLedger { run_id: "run-x".into(), + cap: u64::MAX, + start: start(), spent: 1, }) .unwrap(), @@ -813,12 +825,14 @@ fn the_store_refuses_any_row_that_is_not_a_canonical_v1_record() { ( "a ledger that is not canonical", envelope( - RUN_SPEND_ROW, - Some(RUN_SPEND_SCHEMA), + RUN_LEDGER_ROW, + Some(RUN_LEDGER_SCHEMA), "run-x", [ - rmp_serde::to_vec_named(&RunSpend { + rmp_serde::to_vec_named(&RunLedger { run_id: "run-x".into(), + cap: u64::MAX, + start: start(), spent: 1, }) .unwrap(), @@ -909,7 +923,7 @@ async fn replay_keys_the_request_content_not_the_command_identity() { .unwrap(); } // A fresh genesis names every command anew; the content is what repeats. - let replay = ReplayInferencePort::new("verse-replay", store.records(&run).unwrap()).unwrap(); + let replay = ReplayInferencePort::new("verse-replay", &store, &run, &start()).unwrap(); let under_new_names = |prompt: &str, max: u32| { replay .prepare(request_for(CommandId::new(), prompt, max)) @@ -928,3 +942,263 @@ async fn replay_keys_the_request_content_not_the_command_identity() { assert_eq!(json(&one), json(&reply("one", "r1"))); assert_eq!(json(&two), json(&reply("two", "r2"))); } + +#[tokio::test] +async fn usage_one_token_over_the_bound_faults_and_stores_no_record() { + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let prepared = prepared_for("one over", 1_200); + let bound = bound_of(&prepared); + let inner = ScriptedPort::new(vec![Ok( + reply("edge", "r1").with_usage(usage(bound, 1, None)) + )]); + let port = recording(&inner, &store, "run-edge", bound + 200); + let fault = port.infer(prepared).await.expect_err("one over"); + assert_eq!(fault.class(), InferenceFaultClass::BudgetExhausted); + assert_eq!(port.spent(), bound + 1, "charged in full"); + assert!( + store + .records(&RunId::new("run-edge").unwrap()) + .unwrap() + .is_empty() + ); +} + +/// Every field of the provider request is in the replay key except the two +/// ids that name the call. +#[test] +fn the_replay_key_covers_what_the_request_asks_and_not_what_names_it() { + let base = prepared_for("key", 1_200).invocation.request.clone(); + let digest = |edit: &dyn Fn(&mut CodexProviderRequest)| { + let mut request = base.clone(); + edit(&mut request); + content_digest(&request).unwrap() + }; + let same = digest(&|_| {}); + assert_eq!(same, content_digest(&base).unwrap()); + let changes: [(&str, Box); 10] = [ + ("model", Box::new(|r| r.model.push('x'))), + ("instructions", Box::new(|r| r.instructions.push('x'))), + ( + "tools", + Box::new(|r| { + r.tools.push(CodexToolDefinition { + name: "t".into(), + description: "d".into(), + parameters_json: "{}".into(), + }) + }), + ), + ( + "input", + Box::new(|r| r.input.push(CodexInputItem::UserText { text: "x".into() })), + ), + ( + "max_output_tokens", + Box::new(|r| r.max_output_tokens = Some(1)), + ), + ( + "reasoning_effort", + Box::new(|r| r.reasoning_effort = Some("low".into())), + ), + ( + "service_tier", + Box::new(|r| r.service_tier = Some("flex".into())), + ), + ( + "output_schema_json", + Box::new(|r| r.output_schema_json = Some("{}".into())), + ), + ( + "previous_response_id", + Box::new(|r| r.previous_response_id = Some("p".into())), + ), + ( + "parallel_tool_calls", + Box::new(|r| r.parallel_tool_calls = !r.parallel_tool_calls), + ), + ]; + for (field, change) in changes { + assert_ne!(digest(&*change), same, "{field} is not in the replay key"); + } + assert_eq!(digest(&|r| r.request_id.push('x')), same); + assert_eq!(digest(&|r| r.conversation_id.push('x')), same); +} + +fn open_with( + rows: Vec, +) -> (tempfile::TempDir, Result) { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("calls.redb"); + let mut backing = OwnedRedbMessagePackBackingStore::new(&path).unwrap(); + for row in &rows { + backing.push(row).unwrap(); + } + drop(backing); + let opened = CallRecordStore::open(&path); + (directory, opened) +} + +fn record_row(record: &ProviderCallRecord) -> CultCacheEnvelope { + envelope( + CALL_RECORD_ROW, + Some(CALL_RECORD_SCHEMA), + &row_key(&record.run_id, record.sequence), + rmp_serde::to_vec_named(record).unwrap(), + ) +} + +#[test] +fn a_run_with_sequence_zero_beside_a_hole_is_refused() { + let (_directory, opened) = open_with(vec![ + record_row(&sample_record("run-x", 0)), + record_row(&sample_record("run-x", 2)), + spend_row("run-x", 6), + ]); + assert_eq!(opened.err(), Some(CallRecordError::Corrupt)); +} + +#[test] +fn a_record_charged_nothing_still_needs_its_ledger() { + let mut record = sample_record("run-y", 1); + record.charged = 0; + let (_directory, opened) = open_with(vec![record_row(&record)]); + assert_eq!(opened.err(), Some(CallRecordError::Corrupt)); +} + +#[test] +fn a_record_whose_run_id_is_no_whole_key_segment_is_refused() { + let mut stray = sample_record("run-z/b", 1); + stray.charged = 0; + let (_directory, opened) = open_with(vec![ + record_row(&stray), + record_row(&sample_record("run-z", 1)), + spend_row("run-z", 3), + ]); + assert_eq!(opened.err(), Some(CallRecordError::Corrupt)); + // A neighbouring run id that merely shares the prefix keeps its own records. + let (_directory, opened) = open_with(vec![ + record_row(&sample_record("run-z", 1)), + spend_row("run-z", 3), + spend_row("run-z-b", 0), + ]); + let store = opened.unwrap(); + assert_eq!( + store.records(&RunId::new("run-z").unwrap()).unwrap().len(), + 1 + ); +} + +#[tokio::test] +async fn a_run_reopens_only_under_the_cap_it_began_with() { + let directory = tempfile::tempdir().unwrap(); + let (store, path) = store_in(&directory); + let inner = ScriptedPort::new(Vec::new()); + drop(recording(&inner, &store, "run-cap", 5_000)); + let run = RunId::new("run-cap").unwrap(); + let open = |store: &Arc, cap: u64| { + RecordingInferencePort::open(inner.clone(), store.clone(), run.clone(), cap, &start()) + }; + for other in [4_999, 5_001, u64::MAX] { + assert_eq!( + open(&store, other).err(), + Some(CallRecordError::RunMismatch(RunField::Cap)) + ); + } + // A refused reopen does not hold the run, and the cap outlives the process. + drop(open(&store, 5_000).expect("the same cap reopens")); + drop(store); + let store = Arc::new(CallRecordStore::open(&path).unwrap()); + assert_eq!( + open(&store, 5_001).err(), + Some(CallRecordError::RunMismatch(RunField::Cap)) + ); + open(&store, 5_000).expect("the same cap reopens after a restart"); + let message = format!("{}", CallRecordError::RunMismatch(RunField::Cap)); + assert!(message.contains("token cap") && !message.contains("5000")); +} + +/// A world made by the production path: one genesis command in a fresh world +/// file, its snapshot, and the file's path. +async fn genesis( + directory: &std::path::Path, + command: CommandId, +) -> (crate::WorldSnapshot, std::path::PathBuf) { + let path = directory.join("world.cc"); + let (mailbox, task) = crate::WorldMailbox::open(&path).expect("an empty world"); + mailbox + .create_fixture( + crate::tests::creation(command, "Replayed"), + &crate::tests::auth_principal(crate::tests::owner()), + ) + .await + .expect("a created world"); + let snapshot = mailbox.snapshot().await.unwrap(); + drop(mailbox); + task.await.unwrap(); + (snapshot, path) +} + +async fn reopened(path: &std::path::Path) -> crate::WorldSnapshot { + let (mailbox, task) = crate::WorldMailbox::open(path).expect("a restored world"); + let snapshot = mailbox.snapshot().await.unwrap(); + drop(mailbox); + task.await.unwrap(); + snapshot +} + +#[tokio::test] +async fn replay_runs_against_a_restored_copy_and_refuses_any_other_world_up_front() { + let command = CommandId::new(); + let first = tempfile::tempdir().unwrap(); + let (world, world_file) = genesis(first.path(), command).await; + let began = RunStart::of(&world); + + let directory = tempfile::tempdir().unwrap(); + let (store, _) = store_in(&directory); + let run = RunId::new("run-world").unwrap(); + let inner = ScriptedPort::new(ok_all(&[reply("one", "r1")])); + let live = + RecordingInferencePort::open(inner.clone(), store.clone(), run.clone(), u64::MAX, &began) + .unwrap(); + live.infer(live.prepare(request_for(command, "go", 1_200)).unwrap()) + .await + .unwrap(); + drop(live); + + // A byte copy of the starting world, restored, replays the run. + let restored_dir = tempfile::tempdir().unwrap(); + let restored_file = restored_dir.path().join("world.cc"); + std::fs::copy(&world_file, &restored_file).unwrap(); + let restored = RunStart::of(&reopened(&restored_file).await); + assert_eq!(restored, began); + let replay = ReplayInferencePort::new("verse-replay", &store, &run, &restored) + .expect("a restored copy replays"); + let served = replay + .infer(replay.prepare(request_for(command, "go", 1_200)).unwrap()) + .await + .unwrap(); + assert_eq!(json(&served), json(&reply("one", "r1"))); + + // The same genesis created again is another world and is refused before + // any call, naming the field and nothing it held. + let again = tempfile::tempdir().unwrap(); + let (recreated, _) = genesis(again.path(), command).await; + let refusal = ReplayInferencePort::new("verse-replay", &store, &run, &RunStart::of(&recreated)) + .err() + .expect("a re-created world"); + assert_eq!(refusal, CallRecordError::RunMismatch(RunField::WorldId)); + let text = format!("{refusal} {refusal:?}"); + assert!(!text.contains(&began.world_id) && !text.contains(&began.state_digest)); + + // The same id with another digest is refused too. + let drifted = RunStart { + world_id: began.world_id.clone(), + state_digest: format!("{}x", began.state_digest), + }; + assert_eq!( + ReplayInferencePort::new("verse-replay", &store, &run, &drifted).err(), + Some(CallRecordError::RunMismatch(RunField::StateDigest)) + ); + assert_eq!(inner.calls(), 1, "replay never reached the provider"); +} diff --git a/crates/ghostlight/src/verse/mod.rs b/crates/ghostlight/src/verse/mod.rs index 7c7b6fb..ba0cd08 100644 --- a/crates/ghostlight/src/verse/mod.rs +++ b/crates/ghostlight/src/verse/mod.rs @@ -6,5 +6,5 @@ mod calls; pub use calls::{ CALL_RECORD_SCHEMA, CallRecordError, CallRecordStore, ProviderCallRecord, - RecordingInferencePort, ReplayInferencePort, RunId, + RecordingInferencePort, ReplayInferencePort, RunField, RunId, RunStart, };