Merge ghostlight-verse/verse-run (cuts verse-run r3, lookup, claim, refusals)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
commit
d1111f4c4f
15 changed files with 6580 additions and 86 deletions
12
Cargo.lock
generated
12
Cargo.lock
generated
|
|
@ -730,6 +730,18 @@ dependencies = [
|
|||
"unicode-segmentation",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ghostlight-verse"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"chrono",
|
||||
"ghostlight",
|
||||
"serde_json",
|
||||
"tempfile",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hmac"
|
||||
version = "0.12.1"
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ members = [
|
|||
"crates/ghostlight",
|
||||
"crates/ghostlight-dungeon",
|
||||
"crates/ghostlight-persona-projection",
|
||||
"crates/ghostlight-verse",
|
||||
]
|
||||
resolver = "2"
|
||||
|
||||
|
|
|
|||
15
crates/ghostlight-verse/Cargo.toml
Normal file
15
crates/ghostlight-verse/Cargo.toml
Normal file
|
|
@ -0,0 +1,15 @@
|
|||
[package]
|
||||
name = "ghostlight-verse"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
|
||||
[dependencies]
|
||||
anyhow.workspace = true
|
||||
chrono.workspace = true
|
||||
ghostlight = { path = "../ghostlight" }
|
||||
tokio.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
serde_json.workspace = true
|
||||
tempfile = "3"
|
||||
346
crates/ghostlight-verse/src/main.rs
Normal file
346
crates/ghostlight-verse/src/main.rs
Normal file
|
|
@ -0,0 +1,346 @@
|
|||
//! The verse runner's operator command line.
|
||||
//!
|
||||
//! `genesis` creates the canonical Aetheria verse world in a state root, and
|
||||
//! `run` advances it a given number of ticks under a token cap, printing per
|
||||
//! tick what each faction did and what it cost. A run starts only when this
|
||||
//! command is invoked, and ends when it has done what it was asked.
|
||||
//!
|
||||
//! The state root holds `world.cc`, `controller-work.cc`, `runs.cc` and
|
||||
//! `calls.cc`. The provider key is read from a file at run time and appears
|
||||
//! nowhere else.
|
||||
|
||||
use anyhow::{Context, Result, anyhow, bail};
|
||||
use chrono::{Duration, Utc};
|
||||
use ghostlight::{
|
||||
CommandBody, CommandId, ControllerModels, ControllerOpenError, ControllerRunner, DEFAULT_HOSTED_MODEL_PREFIX,
|
||||
DEFAULT_LOCAL_MODEL_PREFIX, CallRecordStore, CoverBudget, CreateVerseWorldIntent,
|
||||
HostedBinding, Lens, LensWeights, LocalBinding, PrincipalCommandIntent, RecordingInferencePort,
|
||||
RunStart, RunStatus, TickMinutes, VerifiedPrincipalEvidence, VerseRunConfig, VerseRunStore,
|
||||
WorldMailbox, call_run_id, id_text, open_controller_work, open_inference,
|
||||
read_aetheria_catalog, run_manual,
|
||||
};
|
||||
use std::collections::BTreeMap;
|
||||
use std::net::SocketAddr;
|
||||
use std::ops::ControlFlow;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::ExitCode;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
|
||||
/// What a manual run spends per tick on cognition cells. The operator's lever
|
||||
/// on cost is the token cap; the partition always covers every faction.
|
||||
const COVER: CoverBudget = CoverBudget {
|
||||
cells: 8,
|
||||
constituent_cap: 8,
|
||||
urgency_slots: 2,
|
||||
};
|
||||
|
||||
const RUNTIME_ID: &str = "ghostlight-verse";
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> ExitCode {
|
||||
let arguments: Vec<String> = std::env::args().skip(1).collect();
|
||||
match command(arguments).await {
|
||||
Ok(code) => code,
|
||||
Err(error) => {
|
||||
eprintln!("ghostlight-verse: {error:#}");
|
||||
ExitCode::FAILURE
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn command(arguments: Vec<String>) -> Result<ExitCode> {
|
||||
let Some((name, rest)) = arguments.split_first() else {
|
||||
bail!("usage: ghostlight-verse genesis|run [flags]");
|
||||
};
|
||||
match name.as_str() {
|
||||
"genesis" => genesis(Flags::parse(rest, &GENESIS_FLAGS)?).await,
|
||||
"run" => run(Flags::parse(rest, &RUN_FLAGS)?).await,
|
||||
_ => bail!("unknown command, expected genesis or run"),
|
||||
}
|
||||
}
|
||||
|
||||
const GENESIS_FLAGS: [&str; 4] = ["catalog", "aetheria-commit", "state", "owner-hash"];
|
||||
const RUN_FLAGS: [&str; 7] = [
|
||||
"state",
|
||||
"ticks",
|
||||
"token-cap",
|
||||
"tick-span-minutes",
|
||||
"model",
|
||||
"base-url",
|
||||
"key-file",
|
||||
];
|
||||
/// Flags a command may omit.
|
||||
const OPTIONAL: [&str; 1] = ["key-file"];
|
||||
|
||||
/// `--name value` pairs, each named once and from the command's own list.
|
||||
struct Flags(BTreeMap<&'static str, String>);
|
||||
|
||||
impl Flags {
|
||||
fn parse(arguments: &[String], allowed: &[&'static str]) -> Result<Self> {
|
||||
let mut seen = BTreeMap::new();
|
||||
let mut rest = arguments.iter();
|
||||
while let Some(flag) = rest.next() {
|
||||
let Some(name) = flag.strip_prefix("--") else {
|
||||
bail!("expected a --flag, found a bare argument");
|
||||
};
|
||||
let Some(&known) = allowed.iter().find(|candidate| **candidate == name) else {
|
||||
bail!("a flag this command does not take was given");
|
||||
};
|
||||
let Some(value) = rest.next() else {
|
||||
bail!("--{known} needs a value");
|
||||
};
|
||||
if seen.insert(known, value.clone()).is_some() {
|
||||
bail!("--{known} was given twice");
|
||||
}
|
||||
}
|
||||
for &required in allowed {
|
||||
if !OPTIONAL.contains(&required) && !seen.contains_key(required) {
|
||||
bail!("--{required} is required");
|
||||
}
|
||||
}
|
||||
Ok(Self(seen))
|
||||
}
|
||||
|
||||
fn text(&self, name: &str) -> &str {
|
||||
self.0.get(name).map_or("", String::as_str)
|
||||
}
|
||||
|
||||
fn path(&self, name: &str) -> PathBuf {
|
||||
PathBuf::from(self.text(name))
|
||||
}
|
||||
|
||||
fn number<T: std::str::FromStr + PartialOrd + Default>(&self, name: &str) -> Result<T> {
|
||||
match self.text(name).parse::<T>() {
|
||||
Ok(value) if value > T::default() => Ok(value),
|
||||
_ => bail!("--{name} is not a positive integer"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn genesis(flags: Flags) -> Result<ExitCode> {
|
||||
let state = flags.path("state");
|
||||
std::fs::create_dir_all(&state).context("the state directory cannot be created")?;
|
||||
let catalog = read_aetheria_catalog(flags.path("catalog"))
|
||||
.context("the Aetheria catalog cannot be read")?;
|
||||
let factions = catalog.factions.len();
|
||||
let principal = VerifiedPrincipalEvidence::new(flags.text("owner-hash"), Utc::now() + Duration::hours(1));
|
||||
|
||||
let (mailbox, task) = WorldMailbox::open(state.join("world.cc"))
|
||||
.map_err(store_fault("the world file in --state cannot be opened"))?;
|
||||
let receipt = mailbox
|
||||
.create_verse(
|
||||
CreateVerseWorldIntent {
|
||||
id: CommandId::new(),
|
||||
title: "The Aetheria Verse".into(),
|
||||
brief: "The Aetheria verse, simulated from its catalog.".into(),
|
||||
catalog,
|
||||
aetheria_commit: flags.text("aetheria-commit").into(),
|
||||
lens_weights: LensWeights::new(Lens::ALL.into_iter().map(|lens| (lens, 1)).collect()),
|
||||
targets: BTreeMap::new(),
|
||||
},
|
||||
&principal,
|
||||
)
|
||||
.await
|
||||
.context("the verse world cannot be created")?;
|
||||
// A created world is a draft: the owner approves and activates it, so the
|
||||
// factions hold opportunities.
|
||||
for body in [CommandBody::ApproveDraft, CommandBody::ActivateWorld] {
|
||||
let revision = mailbox.snapshot().await.context("the world cannot be read")?.revision;
|
||||
mailbox
|
||||
.submit_principal(
|
||||
PrincipalCommandIntent {
|
||||
id: CommandId::new(),
|
||||
world_id: receipt.world_id,
|
||||
expected_revision: revision,
|
||||
body,
|
||||
},
|
||||
&principal,
|
||||
)
|
||||
.await
|
||||
.context("the world cannot be activated")?;
|
||||
}
|
||||
drop(mailbox);
|
||||
task.await.context("the world file did not close cleanly")?;
|
||||
println!("world {}", id_text(receipt.world_id));
|
||||
println!("factions {factions}");
|
||||
println!("phase Active");
|
||||
Ok(ExitCode::SUCCESS)
|
||||
}
|
||||
|
||||
async fn run(flags: Flags) -> Result<ExitCode> {
|
||||
let state = flags.path("state");
|
||||
let ticks: u32 = flags.number("ticks")?;
|
||||
let cap: u64 = flags.number("token-cap")?;
|
||||
let span: u32 = flags.number("tick-span-minutes")?;
|
||||
let tick_span = TickMinutes::new(span).context("--tick-span-minutes is outside the clock's span")?;
|
||||
if !state.join("world.cc").is_file() {
|
||||
bail!("the state directory holds no world; run genesis first");
|
||||
}
|
||||
|
||||
let model = flags.text("model").to_owned();
|
||||
let models = ControllerModels {
|
||||
projector: model.clone(),
|
||||
persona: model.clone(),
|
||||
interpreter: model.clone(),
|
||||
operational_agent: model.clone(),
|
||||
elaborator: model.clone(),
|
||||
};
|
||||
let inner = open_lane(&flags, &model, &models)?;
|
||||
|
||||
let (mailbox, task) = WorldMailbox::open(state.join("world.cc"))
|
||||
.map_err(store_fault("the world file in --state cannot be opened"))?;
|
||||
let world = mailbox.snapshot().await.context("the world cannot be read")?;
|
||||
let calls = Arc::new(CallRecordStore::open(state.join("calls.cc")).map_err(store_fault("the call records in --state cannot be opened"))?);
|
||||
let runs = VerseRunStore::open(state.join("runs.cc")).map_err(store_fault("the run records in --state cannot be opened"))?;
|
||||
let work = open_controller_work(state.join("controller-work.cc"))
|
||||
.map_err(store_fault("the controller work store in --state cannot be opened"))?;
|
||||
let config = VerseRunConfig {
|
||||
ticks,
|
||||
tick_span,
|
||||
cover: COVER,
|
||||
};
|
||||
let claim = runs
|
||||
.claim(&world, cap, &config)
|
||||
.context("the run cannot be claimed")?;
|
||||
let sequence = claim.sequence();
|
||||
// A start that fails after the claim ends the claimed run, so it binds
|
||||
// nothing: the next claim takes the next number.
|
||||
let started = RecordingInferencePort::open(
|
||||
inner,
|
||||
calls,
|
||||
call_run_id(sequence),
|
||||
cap,
|
||||
&RunStart::of(&world),
|
||||
)
|
||||
.context("the run's call records cannot be opened")
|
||||
.and_then(|port| {
|
||||
let port = Arc::new(port);
|
||||
ControllerRunner::open(mailbox.clone(), port.clone(), work, models)
|
||||
.map(|runner| (port, runner))
|
||||
.context("the controller organ cannot open")
|
||||
});
|
||||
let (port, runner) = match started {
|
||||
Ok(started) => started,
|
||||
Err(error) => {
|
||||
return Err(match runs.abandon(claim, "start_refused") {
|
||||
Ok(()) => error,
|
||||
Err(_) => error.context("the refused run's record could not be ended"),
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
println!(
|
||||
"run {sequence}: {ticks} ticks of {span} minutes, cap {cap} tokens, model {model}"
|
||||
);
|
||||
let interrupted = Arc::new(AtomicBool::new(false));
|
||||
{
|
||||
let interrupted = interrupted.clone();
|
||||
tokio::spawn(async move {
|
||||
if tokio::signal::ctrl_c().await.is_ok() {
|
||||
interrupted.store(true, Ordering::SeqCst);
|
||||
}
|
||||
});
|
||||
}
|
||||
let outcome = run_manual(
|
||||
&mailbox,
|
||||
&runner,
|
||||
&port,
|
||||
&runs,
|
||||
claim,
|
||||
&config,
|
||||
|report| {
|
||||
print!("{report}");
|
||||
if interrupted.load(Ordering::SeqCst) {
|
||||
ControlFlow::Break(())
|
||||
} else {
|
||||
ControlFlow::Continue(())
|
||||
}
|
||||
},
|
||||
)
|
||||
.await
|
||||
.context("the run could not be recorded")?;
|
||||
println!(
|
||||
"status: {}; {} of {} ticks closed",
|
||||
outcome.record.status,
|
||||
outcome.record.ticks_closed(),
|
||||
outcome.record.ticks_requested
|
||||
);
|
||||
drop(runner);
|
||||
drop(port);
|
||||
drop(mailbox);
|
||||
task.await.context("the world file did not close cleanly")?;
|
||||
Ok(match outcome.record.status {
|
||||
RunStatus::Failed { .. } => ExitCode::FAILURE,
|
||||
_ => ExitCode::SUCCESS,
|
||||
})
|
||||
}
|
||||
|
||||
/// The inference lane the model's prefix names. A hosted model needs an https
|
||||
/// origin and a key file; a local one reaches a loopback server and needs
|
||||
/// neither, so a key file beside it is refused.
|
||||
fn open_lane(
|
||||
flags: &Flags,
|
||||
model: &str,
|
||||
models: &ControllerModels,
|
||||
) -> Result<Arc<dyn ghostlight::InferencePort>> {
|
||||
let (mut hosted, mut local) = (None, None);
|
||||
if model.starts_with(DEFAULT_HOSTED_MODEL_PREFIX) {
|
||||
let Some(key) = flags.0.get("key-file") else {
|
||||
bail!("a hosted model needs --key-file");
|
||||
};
|
||||
hosted = Some(HostedBinding {
|
||||
base_url: flags.text("base-url").to_owned(),
|
||||
key_path: Path::new(key).to_owned(),
|
||||
model_prefix: DEFAULT_HOSTED_MODEL_PREFIX.into(),
|
||||
caller_runtime_id: RUNTIME_ID.into(),
|
||||
});
|
||||
} else if model.starts_with(DEFAULT_LOCAL_MODEL_PREFIX) {
|
||||
if flags.0.contains_key("key-file") {
|
||||
bail!("a local model takes no --key-file");
|
||||
}
|
||||
let endpoint: SocketAddr = flags
|
||||
.text("base-url")
|
||||
.strip_prefix("http://")
|
||||
.and_then(|address| address.trim_end_matches('/').parse().ok())
|
||||
.context("--base-url for a local model is not http://<address>:<port>")?;
|
||||
local = Some(LocalBinding {
|
||||
endpoint,
|
||||
model_prefix: DEFAULT_LOCAL_MODEL_PREFIX.into(),
|
||||
caller_runtime_id: RUNTIME_ID.into(),
|
||||
});
|
||||
} else {
|
||||
bail!("--model names no inference lane (expected a hosted/ or local/ prefix)");
|
||||
}
|
||||
open_inference(None, None, local, hosted, &models.each())
|
||||
.map_err(|error| anyhow!(lane_fault(&error)))
|
||||
}
|
||||
|
||||
/// A store that cannot open says which one and that it is under `--state`,
|
||||
/// never the path or the store's own text, which carries the path.
|
||||
fn store_fault<E>(message: &'static str) -> impl FnOnce(E) -> anyhow::Error {
|
||||
move |_| anyhow!(message)
|
||||
}
|
||||
|
||||
/// What a lane that cannot open says: the flag or the kind of fault, never a
|
||||
/// value the operator passed.
|
||||
fn lane_fault(error: &ControllerOpenError) -> &'static str {
|
||||
match error {
|
||||
ControllerOpenError::InvalidModels => "--model is not an exact nonempty identifier",
|
||||
ControllerOpenError::Connector(_) => "the connector lane cannot open",
|
||||
ControllerOpenError::WorkStore(_) => "the controller work journal cannot open",
|
||||
ControllerOpenError::UnroutableModel { .. } => "--model is claimed by no inference lane",
|
||||
ControllerOpenError::SdkSidecarMissing { .. } => "the SDK sidecar entry is not a file",
|
||||
ControllerOpenError::LocalEndpointNotLoopback { .. } => {
|
||||
"--base-url for a local model must be a loopback address"
|
||||
}
|
||||
ControllerOpenError::HostedEndpointNotHttps => {
|
||||
"--base-url for a hosted model must be an https URL"
|
||||
}
|
||||
ControllerOpenError::HostedKeyUnreadable => {
|
||||
"--key-file is unreadable, empty, or not one run of visible ASCII"
|
||||
}
|
||||
ControllerOpenError::SharedModelPrefix { .. } => "two inference lanes claim one model prefix",
|
||||
ControllerOpenError::EmptyModelPrefix { .. } => "an inference lane's model prefix is empty",
|
||||
}
|
||||
}
|
||||
445
crates/ghostlight-verse/tests/verse_binary.rs
Normal file
445
crates/ghostlight-verse/tests/verse_binary.rs
Normal file
|
|
@ -0,0 +1,445 @@
|
|||
//! The `ghostlight-verse` command against a loopback scripted responder. No
|
||||
//! real model is called and no key file is read: the responder answers the
|
||||
//! OpenAI-compatible chat-completions request the local lane sends.
|
||||
|
||||
use serde_json::{Value, json};
|
||||
use std::io::{Read, Write};
|
||||
use std::net::{SocketAddr, TcpListener, TcpStream};
|
||||
use std::path::Path;
|
||||
use std::process::{Command, Output};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
const BINARY: &str = env!("CARGO_BIN_EXE_ghostlight-verse");
|
||||
const CATALOG: &str = concat!(
|
||||
env!("CARGO_MANIFEST_DIR"),
|
||||
"/../ghostlight/tests/fixtures/aetheria-verse-catalog.cc"
|
||||
);
|
||||
const COMMIT: &str = "92b82e9aa5bb422c568c4bfc85a8f641426a41fc";
|
||||
/// What the responder reports each call cost: 120 prompt and 30 completion.
|
||||
const PER_CALL: u64 = 150;
|
||||
|
||||
fn verse(arguments: &[&str]) -> Output {
|
||||
Command::new(BINARY).args(arguments).output().unwrap()
|
||||
}
|
||||
|
||||
fn stdout(output: &Output) -> String {
|
||||
String::from_utf8_lossy(&output.stdout).into_owned()
|
||||
}
|
||||
|
||||
fn stderr(output: &Output) -> String {
|
||||
String::from_utf8_lossy(&output.stderr).into_owned()
|
||||
}
|
||||
|
||||
fn genesis(state: &Path) -> Output {
|
||||
verse(&[
|
||||
"genesis",
|
||||
"--catalog",
|
||||
CATALOG,
|
||||
"--aetheria-commit",
|
||||
COMMIT,
|
||||
"--state",
|
||||
state.to_str().unwrap(),
|
||||
"--owner-hash",
|
||||
"verse-owner",
|
||||
])
|
||||
}
|
||||
|
||||
/// Answers every chat-completions request with one tool call: the faction the
|
||||
/// prompt names speaks its own line. Counts the requests it served.
|
||||
struct Responder {
|
||||
address: SocketAddr,
|
||||
served: Arc<AtomicUsize>,
|
||||
}
|
||||
|
||||
impl Responder {
|
||||
fn start() -> Self {
|
||||
Self::answering(200)
|
||||
}
|
||||
|
||||
/// A provider that refuses every call with this status.
|
||||
fn refusing(status: u16) -> Self {
|
||||
Self::answering(status)
|
||||
}
|
||||
|
||||
fn answering(status: u16) -> Self {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||
let address = listener.local_addr().unwrap();
|
||||
let served = Arc::new(AtomicUsize::new(0));
|
||||
let counter = served.clone();
|
||||
std::thread::spawn(move || {
|
||||
for stream in listener.incoming() {
|
||||
let Ok(stream) = stream else { return };
|
||||
let counter = counter.clone();
|
||||
std::thread::spawn(move || serve(stream, &counter, status));
|
||||
}
|
||||
});
|
||||
Self { address, served }
|
||||
}
|
||||
|
||||
fn url(&self) -> String {
|
||||
format!("http://{}", self.address)
|
||||
}
|
||||
}
|
||||
|
||||
fn serve(mut stream: TcpStream, served: &AtomicUsize, status: u16) {
|
||||
let mut raw = Vec::new();
|
||||
let mut chunk = [0u8; 4096];
|
||||
let (head_end, length) = loop {
|
||||
let read = stream.read(&mut chunk).unwrap_or(0);
|
||||
if read == 0 {
|
||||
return;
|
||||
}
|
||||
raw.extend_from_slice(&chunk[..read]);
|
||||
if let Some(end) = raw.windows(4).position(|window| window == b"\r\n\r\n") {
|
||||
let head = String::from_utf8_lossy(&raw[..end]).to_ascii_lowercase();
|
||||
let length = head
|
||||
.lines()
|
||||
.find_map(|line| line.strip_prefix("content-length:"))
|
||||
.and_then(|value| value.trim().parse::<usize>().ok())
|
||||
.unwrap_or(0);
|
||||
break (end + 4, length);
|
||||
}
|
||||
};
|
||||
while raw.len() < head_end + length {
|
||||
let read = stream.read(&mut chunk).unwrap_or(0);
|
||||
if read == 0 {
|
||||
return;
|
||||
}
|
||||
raw.extend_from_slice(&chunk[..read]);
|
||||
}
|
||||
let request: Value = serde_json::from_slice(&raw[head_end..head_end + length]).unwrap();
|
||||
let text = request.to_string();
|
||||
let faction = ["Adrasteia", "Brannoch"]
|
||||
.into_iter()
|
||||
.find(|name| text.contains(&format!("Decision owner:\\n{name}\\n")))
|
||||
.expect("a request names its faction");
|
||||
let sequence = served.fetch_add(1, Ordering::SeqCst);
|
||||
let reply = json!({
|
||||
"id": format!("reply-{sequence}"),
|
||||
"choices": [{
|
||||
"message": {
|
||||
"role": "assistant",
|
||||
"content": null,
|
||||
"tool_calls": [{
|
||||
"id": format!("call-{sequence}"),
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "speak",
|
||||
"arguments": json!({ "text": format!("{faction} holds the line.") }).to_string(),
|
||||
},
|
||||
}],
|
||||
},
|
||||
"finish_reason": "tool_calls",
|
||||
}],
|
||||
"usage": { "prompt_tokens": 120, "completion_tokens": 30 },
|
||||
})
|
||||
.to_string();
|
||||
let response = format!(
|
||||
"HTTP/1.1 {status} Scripted\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{reply}",
|
||||
reply.len()
|
||||
);
|
||||
let _ = stream.write_all(response.as_bytes());
|
||||
}
|
||||
|
||||
fn run(state: &Path, responder: &Responder, ticks: &str, cap: &str) -> Output {
|
||||
verse(&[
|
||||
"run",
|
||||
"--state",
|
||||
state.to_str().unwrap(),
|
||||
"--ticks",
|
||||
ticks,
|
||||
"--token-cap",
|
||||
cap,
|
||||
"--tick-span-minutes",
|
||||
"60",
|
||||
"--model",
|
||||
"local/scripted",
|
||||
"--base-url",
|
||||
&responder.url(),
|
||||
])
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn binary_run_prints_acts_and_spend() {
|
||||
let state = tempfile::tempdir().unwrap();
|
||||
let made = genesis(state.path());
|
||||
assert!(made.status.success(), "{}", stderr(&made));
|
||||
let made = stdout(&made);
|
||||
assert!(made.contains("factions 2\n") && made.contains("phase Active\n"), "{made}");
|
||||
assert!(state.path().join("world.cc").is_file());
|
||||
|
||||
let responder = Responder::start();
|
||||
let output = run(state.path(), &responder, "2", "1000000");
|
||||
assert!(output.status.success(), "{}", stderr(&output));
|
||||
let text = stdout(&output);
|
||||
|
||||
// Two ticks, each with both factions named by verb, and the spend as it
|
||||
// accrues against the cap.
|
||||
assert!(text.starts_with("run 1: 2 ticks of 60 minutes, cap 1000000 tokens"), "{text}");
|
||||
for tick in 0..2u64 {
|
||||
let spent = PER_CALL * 2 * (tick + 1);
|
||||
assert!(
|
||||
text.contains(&format!("tick {tick} closed: 2 cells, 2 acts, 0 refused\n")),
|
||||
"{text}"
|
||||
);
|
||||
assert!(
|
||||
text.contains(&format!(
|
||||
" tokens: {} this tick, {spent} spent, {} remaining\n",
|
||||
PER_CALL * 2,
|
||||
1_000_000 - spent
|
||||
)),
|
||||
"{text}"
|
||||
);
|
||||
}
|
||||
for faction in ["Adrasteia", "Brannoch"] {
|
||||
assert_eq!(
|
||||
text.matches(&format!(" {faction} speak [")).count(),
|
||||
2,
|
||||
"{text}"
|
||||
);
|
||||
assert!(text.contains(&format!("says \"{faction} holds the line.\"")), "{text}");
|
||||
}
|
||||
assert!(text.ends_with("status: Completed; 2 of 2 ticks closed\n"), "{text}");
|
||||
assert_eq!(responder.served.load(Ordering::SeqCst), 4);
|
||||
|
||||
// The state root holds each store the run wrote.
|
||||
for file in ["world.cc", "controller-work.cc", "runs.cc", "calls.cc"] {
|
||||
assert!(state.path().join(file).is_file(), "{file} was not written");
|
||||
}
|
||||
|
||||
// A second run is the next in the world and carries on from its clock.
|
||||
let again = run(state.path(), &responder, "1", "1000000");
|
||||
assert!(again.status.success(), "{}", stderr(&again));
|
||||
let text = stdout(&again);
|
||||
assert!(text.starts_with("run 2: 1 ticks of 60 minutes"), "{text}");
|
||||
assert!(text.contains("tick 2 closed: 2 cells, 2 acts, 0 refused\n"), "{text}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_cap_the_first_call_could_pass_ends_the_run_with_nothing_spent() {
|
||||
let state = tempfile::tempdir().unwrap();
|
||||
assert!(genesis(state.path()).status.success());
|
||||
let responder = Responder::start();
|
||||
let output = run(state.path(), &responder, "3", "1000");
|
||||
assert!(output.status.success(), "{}", stderr(&output));
|
||||
let text = stdout(&output);
|
||||
assert!(text.contains("tick 0 open: "), "{text}");
|
||||
assert!(text.contains("tokens: 0 this tick, 0 spent, 1000 remaining"), "{text}");
|
||||
assert!(text.ends_with("status: BudgetExhausted; 0 of 3 ticks closed\n"), "{text}");
|
||||
assert_eq!(responder.served.load(Ordering::SeqCst), 0, "no call was made");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_run_that_fails_says_so_and_exits_nonzero() {
|
||||
let state = tempfile::tempdir().unwrap();
|
||||
assert!(genesis(state.path()).status.success());
|
||||
let responder = Responder::refusing(500);
|
||||
let output = run(state.path(), &responder, "2", "1000000");
|
||||
assert!(!output.status.success());
|
||||
let text = stdout(&output);
|
||||
assert!(text.contains("tick 0 open: "), "{text}");
|
||||
assert!(
|
||||
text.ends_with("status: Failed (inference_recovery_required); 0 of 2 ticks closed\n"),
|
||||
"{text}"
|
||||
);
|
||||
assert_eq!(
|
||||
responder.served.load(Ordering::SeqCst),
|
||||
1,
|
||||
"the run stopped at the first refusal"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_command_refuses_what_it_cannot_run_and_never_echoes_a_value() {
|
||||
let state = tempfile::tempdir().unwrap();
|
||||
let responder = Responder::start();
|
||||
|
||||
// No world yet.
|
||||
let output = run(state.path(), &responder, "1", "1000");
|
||||
assert!(!output.status.success());
|
||||
assert!(stderr(&output).contains("run genesis first"), "{}", stderr(&output));
|
||||
|
||||
assert!(genesis(state.path()).status.success());
|
||||
// A second genesis does not overwrite the first.
|
||||
assert!(!genesis(state.path()).status.success());
|
||||
|
||||
let secret = "sk-not-a-real-key-0000";
|
||||
let key = state.path().join(secret);
|
||||
let cases: [(Vec<&str>, &str); 9] = [
|
||||
(vec![], "usage"),
|
||||
(vec!["fly"], "unknown command"),
|
||||
(
|
||||
vec!["run", "--state", state.path().to_str().unwrap(), "--ticks", "0"],
|
||||
"--token-cap is required",
|
||||
),
|
||||
(
|
||||
vec![
|
||||
"run", "--state", state.path().to_str().unwrap(), "--ticks", "many",
|
||||
"--token-cap", "1", "--tick-span-minutes", "60", "--model", "local/x",
|
||||
"--base-url", "http://127.0.0.1:1",
|
||||
],
|
||||
"--ticks is not a positive integer",
|
||||
),
|
||||
(
|
||||
vec![
|
||||
"run", "--state", state.path().to_str().unwrap(), "--ticks", "0",
|
||||
"--token-cap", "1", "--tick-span-minutes", "60", "--model", "local/x",
|
||||
"--base-url", "http://127.0.0.1:1",
|
||||
],
|
||||
"--ticks is not a positive integer",
|
||||
),
|
||||
(
|
||||
vec![
|
||||
"run", "--state", state.path().to_str().unwrap(), "--ticks", "1",
|
||||
"--token-cap", "0", "--tick-span-minutes", "60", "--model", "local/x",
|
||||
"--base-url", "http://127.0.0.1:1",
|
||||
],
|
||||
"--token-cap is not a positive integer",
|
||||
),
|
||||
(
|
||||
vec![
|
||||
"run", "--state", state.path().to_str().unwrap(), "--ticks", "1",
|
||||
"--token-cap", "1", "--tick-span-minutes", "60", "--model", "gpt-4",
|
||||
"--base-url", "http://127.0.0.1:1",
|
||||
],
|
||||
"--model names no inference lane",
|
||||
),
|
||||
(
|
||||
// A hosted model needs its key file; a missing one is named by flag.
|
||||
vec![
|
||||
"run", "--state", state.path().to_str().unwrap(), "--ticks", "1",
|
||||
"--token-cap", "1", "--tick-span-minutes", "60", "--model", "hosted/x",
|
||||
"--base-url", "https://api.example.test", "--key-file", key.to_str().unwrap(),
|
||||
],
|
||||
"--key-file is unreadable",
|
||||
),
|
||||
(
|
||||
vec![
|
||||
"run", "--state", state.path().to_str().unwrap(), "--ticks", "1",
|
||||
"--token-cap", "1", "--tick-span-minutes", "60", "--model", "local/x",
|
||||
"--base-url", "http://127.0.0.1:1", "--key-file", key.to_str().unwrap(),
|
||||
],
|
||||
"a local model takes no --key-file",
|
||||
),
|
||||
];
|
||||
for (arguments, expected) in cases {
|
||||
let output = verse(&arguments);
|
||||
assert!(!output.status.success(), "{arguments:?} succeeded");
|
||||
let error = stderr(&output);
|
||||
assert!(error.contains(expected), "{arguments:?}: {error}");
|
||||
assert!(!error.contains(secret), "{arguments:?} echoed a path: {error}");
|
||||
assert!(!error.contains("many"), "{arguments:?} echoed a value: {error}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_non_loopback_local_endpoint_is_refused_without_echoing_it() {
|
||||
let state = tempfile::tempdir().unwrap();
|
||||
assert!(genesis(state.path()).status.success());
|
||||
let output = verse(&[
|
||||
"run",
|
||||
"--state",
|
||||
state.path().to_str().unwrap(),
|
||||
"--ticks",
|
||||
"1",
|
||||
"--token-cap",
|
||||
"1000",
|
||||
"--tick-span-minutes",
|
||||
"60",
|
||||
"--model",
|
||||
"local/x",
|
||||
"--base-url",
|
||||
"http://10.77.0.99:4242",
|
||||
]);
|
||||
assert!(!output.status.success());
|
||||
let error = stderr(&output);
|
||||
assert!(error.contains("--base-url"), "{error}");
|
||||
assert!(!error.contains("10.77.0.99"), "the endpoint was echoed: {error}");
|
||||
// The lane failed before any run was claimed.
|
||||
assert!(!stdout(&output).contains("run 1"), "{}", stdout(&output));
|
||||
}
|
||||
|
||||
/// A start refused after its claim ends the claimed record Failed, so the run
|
||||
/// binds nothing. Reached from the command by removing the run records while
|
||||
/// the call ledger still holds run 1 under another cap.
|
||||
#[test]
|
||||
fn a_start_the_ledger_refuses_ends_its_claimed_record_failed() {
|
||||
let parent = tempfile::tempdir().unwrap();
|
||||
let state = parent.path().join("state");
|
||||
std::fs::create_dir_all(&state).unwrap();
|
||||
assert!(genesis(&state).status.success());
|
||||
let responder = Responder::start();
|
||||
let first = run(&state, &responder, "1", "1000000");
|
||||
assert!(first.status.success(), "{}", stderr(&first));
|
||||
std::fs::remove_file(state.join("runs.cc")).unwrap();
|
||||
|
||||
let refused = run(&state, &responder, "1", "500000");
|
||||
assert!(!refused.status.success());
|
||||
|
||||
let world = {
|
||||
let runtime = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.unwrap();
|
||||
let _entered = runtime.enter();
|
||||
let (mailbox, task) = ghostlight::WorldMailbox::open(state.join("world.cc")).unwrap();
|
||||
let world = runtime.block_on(mailbox.snapshot()).unwrap().world_id;
|
||||
drop(mailbox);
|
||||
runtime.block_on(task).unwrap();
|
||||
world
|
||||
};
|
||||
let runs = ghostlight::VerseRunStore::open(state.join("runs.cc"))
|
||||
.unwrap()
|
||||
.runs(world)
|
||||
.unwrap();
|
||||
assert_eq!(runs.len(), 1);
|
||||
assert_eq!(
|
||||
runs[0].status,
|
||||
ghostlight::RunStatus::Failed {
|
||||
reason: "start_refused".into()
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
/// A store that cannot open names the flag and the fault, never the path.
|
||||
#[test]
|
||||
fn a_store_that_cannot_open_never_echoes_the_state_path() {
|
||||
let responder = Responder::start();
|
||||
for file in ["world.cc", "controller-work.cc"] {
|
||||
for held in [false, true] {
|
||||
let parent = tempfile::tempdir().unwrap();
|
||||
let state = parent.path().join("distinctivestatename");
|
||||
std::fs::create_dir_all(&state).unwrap();
|
||||
assert!(genesis(&state).status.success());
|
||||
let first = run(&state, &responder, "1", "1000000");
|
||||
assert!(first.status.success(), "{}", stderr(&first));
|
||||
|
||||
let target = state.join(file);
|
||||
let _guard: Box<dyn std::any::Any> = if !held {
|
||||
std::fs::write(&target, b"not a database").unwrap();
|
||||
Box::new(())
|
||||
} else if file == "world.cc" {
|
||||
let runtime = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.unwrap();
|
||||
let entered = runtime.enter();
|
||||
let opened = ghostlight::WorldMailbox::open(&target).unwrap();
|
||||
drop(entered);
|
||||
Box::new((runtime, opened))
|
||||
} else {
|
||||
Box::new(ghostlight::open_controller_work(&target).unwrap())
|
||||
};
|
||||
|
||||
let output = run(&state, &responder, "1", "1000000");
|
||||
assert!(!output.status.success(), "{file} held={held}");
|
||||
let error = stderr(&output);
|
||||
assert!(error.contains("--state"), "{file} held={held}: {error}");
|
||||
assert!(
|
||||
!error.contains("distinctivestatename"),
|
||||
"{file} held={held} echoed the path: {error}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -469,7 +469,7 @@ mod tests {
|
|||
AuthorityTargetRef, CallerId, CausalBoundary, CommandBody, CommandEnvelope, CommitmentKind,
|
||||
ComponentOp, CoverBudget, DecisionOpportunity, DecisionScope, Declaration, DraftHandle,
|
||||
EntityDeclaration, EntityId, EntityKind, EvidenceRef, FactDeclaration, FactStandingRef,
|
||||
FictionalMinutes, JurisdictionKey, NewController, PatchGround, Quantity, Ref,
|
||||
FictionalMinutes, JurisdictionKey, NewController, PatchGround, Quantity, Ref, TickIndex,
|
||||
RouteDeclaration, Statement, SubjectDeclaration, SubjectId, SubjectKind, SystemCapability,
|
||||
WorldEffect, WorldId, WorldKernel, WorldMailbox, WorldScaleIntentRef, agency_graph,
|
||||
commit_digest, derive_boundaries, derive_cover, derive_opportunities,
|
||||
|
|
@ -736,8 +736,7 @@ mod tests {
|
|||
assert!(!graph.subjects.contains(&mirror.first));
|
||||
let cover = derive_cover(
|
||||
snapshot.world_id,
|
||||
snapshot.now,
|
||||
30,
|
||||
TickIndex(0),
|
||||
&opportunities,
|
||||
&graph,
|
||||
CoverBudget {
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -11,7 +11,6 @@
|
|||
//! The cover is a partition, never a filter: every active subject is in exactly
|
||||
//! one cell every tick. The budget decides at what resolution, not whether.
|
||||
|
||||
use super::clock::FictionalMinutes;
|
||||
use super::{ControllerMode, DecisionOpportunity, SubjectId, WorldId};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sha2::{Digest, Sha256};
|
||||
|
|
@ -29,25 +28,18 @@ pub(crate) const AGENCY_STAR_THRESHOLD: usize = 16;
|
|||
/// the name, not by a second namespace.
|
||||
const COVER_NAMESPACE: &str = "ghostlight.cover.v1";
|
||||
|
||||
/// Monotone tick index. Derived from the world clock, never stored. It is
|
||||
/// threaded into every derived command id, so a resumed tick over the same
|
||||
/// snapshot re-derives the same ids and the kernel's idempotency ledger answers
|
||||
/// with the original receipts instead of double-committing.
|
||||
/// A tick's ordinal: the number of clock advances the world had committed when
|
||||
/// the tick opened. The driver reads it from the world and freezes it in the
|
||||
/// tick document; it is never derived from `now`, so it does not depend on the
|
||||
/// span. It is threaded into every derived command id, so a resumed tick
|
||||
/// re-submits the same ids and the kernel's idempotency ledger answers with the
|
||||
/// original receipts instead of double-committing.
|
||||
#[derive(
|
||||
Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, Default,
|
||||
)]
|
||||
#[serde(transparent)]
|
||||
pub struct TickIndex(pub u64);
|
||||
|
||||
impl TickIndex {
|
||||
/// Floor division by the span one `AdvanceTime` carries. The driver submits
|
||||
/// the clock *after* the cells run, so every cell in one tick sees the same
|
||||
/// `now` and therefore the same index.
|
||||
pub(crate) fn of(now: FictionalMinutes, tick_minutes: u32) -> Self {
|
||||
Self(now.0 / u64::from(tick_minutes.max(1)))
|
||||
}
|
||||
}
|
||||
|
||||
/// How a subject was represented this tick. Derived from the cell, carried
|
||||
/// nowhere in world state: representation is a compute budget, not world truth.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
|
|
@ -254,7 +246,7 @@ impl AgencyGraph {
|
|||
}
|
||||
}
|
||||
|
||||
/// The partition. Pure over `(world, now, opportunities, graph, budget)`.
|
||||
/// The partition. Pure over `(world, tick, opportunities, graph, budget)`.
|
||||
///
|
||||
/// `opportunities` arrives in the kernel's one attention order (pressure, then
|
||||
/// attention debt, then id) and is never re-sorted and never filtered here: a
|
||||
|
|
@ -262,13 +254,11 @@ impl AgencyGraph {
|
|||
/// is a fact about cognition rather than a scheduling preference.
|
||||
pub fn derive_cover(
|
||||
world: WorldId,
|
||||
now: FictionalMinutes,
|
||||
tick_minutes: u32,
|
||||
tick: TickIndex,
|
||||
opportunities: &[DecisionOpportunity],
|
||||
graph: &AgencyGraph,
|
||||
budget: CoverBudget,
|
||||
) -> Cover {
|
||||
let tick = TickIndex::of(now, tick_minutes);
|
||||
let active: Vec<&DecisionOpportunity> = opportunities
|
||||
.iter()
|
||||
.filter(|opportunity| opportunity.controller_mode != ControllerMode::Human)
|
||||
|
|
@ -601,8 +591,7 @@ mod tests {
|
|||
) -> Cover {
|
||||
derive_cover(
|
||||
WorldId::nil_for_test(),
|
||||
FictionalMinutes(tick * 60),
|
||||
60,
|
||||
TickIndex(tick),
|
||||
opportunities,
|
||||
graph,
|
||||
budget,
|
||||
|
|
|
|||
|
|
@ -181,6 +181,11 @@ impl WorldJournal {
|
|||
self.commits.get(&command_id)
|
||||
}
|
||||
|
||||
/// Every commit the journal holds, in no particular order.
|
||||
pub(super) fn commits(&self) -> impl Iterator<Item = &WorldCommit> {
|
||||
self.commits.values()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(super) fn commit_count(&self) -> usize {
|
||||
self.commits.len()
|
||||
|
|
|
|||
|
|
@ -85,6 +85,7 @@ pub use consumer::{
|
|||
};
|
||||
pub use controllers::{
|
||||
CONTROLLER_WORK_SCHEMA, CellRun, ConnectorBinding, ControllerError, ControllerModels,
|
||||
ControllerOpenError,
|
||||
ControllerNeed, ControllerPendingReason, ControllerRunner, ControllerWork,
|
||||
ControllerWorkCustody, ControllerWorkLookup, ControllerWorkStore, ControllerWorkStoreError,
|
||||
ControllerWorkWrite,
|
||||
|
|
@ -145,7 +146,10 @@ pub use table::{
|
|||
pub use vault::{VaultError, VaultEvidenceSource};
|
||||
pub use verse::{
|
||||
CALL_RECORD_SCHEMA, CallRecordError, CallRecordStore, CreateVerseWorldIntent, ProviderCallRecord,
|
||||
RecordingInferencePort, ReplayInferencePort, RunField, RunId, RunStart, VerseGenesisError,
|
||||
RUN_RECORD_SCHEMA, RecordedCoverBudget, RecordingInferencePort, Refusal, RefusalReason, RunClaim,
|
||||
ReplayInferencePort, RunField, RunId, RunOutcome, RunStart, RunStatus, TickCheckpoint,
|
||||
TickReport, UnattributedCall, VerseGenesisError, VerseRunConfig, VerseRunError, VerseRunRecord, VerseRunStore,
|
||||
call_run_id, run_manual,
|
||||
};
|
||||
use patch::{EdgeRecord, EntityRecord, LedgerDelta, ResolvedOp, ResolvedPatch};
|
||||
|
||||
|
|
@ -1275,6 +1279,21 @@ pub struct OperatorEvent {
|
|||
pub speaker_label: String,
|
||||
pub speech: Option<Statement>,
|
||||
pub display: Option<Statement>,
|
||||
/// The affordance kind the act exercised.
|
||||
pub verb: String,
|
||||
/// How the world's grammar renders that verb; none in a world with no
|
||||
/// grammar.
|
||||
pub render_path: Option<RenderPath>,
|
||||
/// What the act bound each of the verb's roles to, in the invocation's
|
||||
/// order.
|
||||
pub roles: Vec<BoundRole>,
|
||||
}
|
||||
|
||||
/// One role an act bound, with the label of what it was bound to.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct BoundRole {
|
||||
pub role: String,
|
||||
pub target: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
|
|
@ -1565,7 +1584,43 @@ impl WorldKernel {
|
|||
/// no consumer may read in full would be a loaded gun with a comment on it.
|
||||
fn operator_log(&self) -> Result<Vec<OperatorEvent>, KernelError> {
|
||||
self.journal.ensure_healthy()?;
|
||||
operator_log(&self.state)
|
||||
// An event records what the world did; the invocation, which names the
|
||||
// bound roles, lives on the commit that produced it.
|
||||
let invocations: BTreeMap<EventId, &DecisionInvocation> = self
|
||||
.journal
|
||||
.commits()
|
||||
.filter_map(|commit| match &commit.effect {
|
||||
WorldEffect::DecisionExercised {
|
||||
invocation, event, ..
|
||||
} => Some((event.id, invocation)),
|
||||
_ => None,
|
||||
})
|
||||
.collect();
|
||||
operator_log(&self.state, &invocations)
|
||||
}
|
||||
|
||||
/// The tick driver's opportunity list, in the same order a snapshot
|
||||
/// carries it. Separate from `snapshot` for the reason `agency_graph` is:
|
||||
/// only the tick driver reads it.
|
||||
fn opportunities(&self) -> Result<Vec<DecisionOpportunity>, KernelError> {
|
||||
self.journal.ensure_healthy()?;
|
||||
Ok(order_opportunities(
|
||||
&self.state,
|
||||
derive_opportunities(&self.state)?,
|
||||
))
|
||||
}
|
||||
|
||||
/// How many clock advances the world has committed: the next tick's
|
||||
/// ordinal, held by the world and read from its journal. Separate from
|
||||
/// `snapshot` for the reason `opportunities` is: only the tick driver
|
||||
/// reads it.
|
||||
fn clock_advances(&self) -> Result<u64, KernelError> {
|
||||
self.journal.ensure_healthy()?;
|
||||
Ok(self
|
||||
.journal
|
||||
.commits()
|
||||
.filter(|commit| matches!(commit.effect, WorldEffect::TimeAdvanced { .. }))
|
||||
.count() as u64)
|
||||
}
|
||||
|
||||
/// The scheduler's adjacency projection. Separate from `snapshot` for the
|
||||
|
|
@ -3740,7 +3795,10 @@ fn scope_components(state: &WorldState, subject_id: SubjectId) -> ScopeComponent
|
|||
/// The human operator's story feed, derived from the causal record. The events
|
||||
/// log is not a perception surface: `knowledge` is the sole answer to what a
|
||||
/// subject may act on and what its projection may contain.
|
||||
fn operator_log(state: &WorldState) -> Result<Vec<OperatorEvent>, KernelError> {
|
||||
fn operator_log(
|
||||
state: &WorldState,
|
||||
invocations: &BTreeMap<EventId, &DecisionInvocation>,
|
||||
) -> Result<Vec<OperatorEvent>, KernelError> {
|
||||
state
|
||||
.events
|
||||
.iter()
|
||||
|
|
@ -3749,6 +3807,30 @@ fn operator_log(state: &WorldState) -> Result<Vec<OperatorEvent>, KernelError> {
|
|||
.subjects
|
||||
.get(&event.scope.subject_id)
|
||||
.ok_or_else(|| KernelError::Invariant("event names no live subject".into()))?;
|
||||
let invocation = invocations.get(&event.id).ok_or_else(|| {
|
||||
KernelError::Invariant("event has no committed invocation".into())
|
||||
})?;
|
||||
let entry = state
|
||||
.affordance_catalog
|
||||
.get(&event.affordance)
|
||||
.ok_or_else(|| KernelError::Invariant("event names no catalog entry".into()))?;
|
||||
let roles = invocation
|
||||
.bindings
|
||||
.iter()
|
||||
.map(|binding| {
|
||||
let label = match binding.target {
|
||||
Target::Subject(id) => state.subjects.get(&id).map(|s| s.label.clone()),
|
||||
Target::Entity(id) => state.entities.get(&id).map(|e| e.label.clone()),
|
||||
Target::Edge(id) => state.edges.contains_key(&id).then(|| "edge".to_owned()),
|
||||
};
|
||||
label
|
||||
.map(|target| BoundRole {
|
||||
role: binding.role.0.clone(),
|
||||
target,
|
||||
})
|
||||
.ok_or_else(|| KernelError::Invariant("event binds no live referent".into()))
|
||||
})
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
Ok(OperatorEvent {
|
||||
revision: event.revision,
|
||||
speaker: event.scope.subject_id,
|
||||
|
|
@ -3761,6 +3843,13 @@ fn operator_log(state: &WorldState) -> Result<Vec<OperatorEvent>, KernelError> {
|
|||
.display
|
||||
.and_then(|fact| state.facts.get(&fact))
|
||||
.map(|record| record.statement.clone()),
|
||||
verb: entry.kind.0.clone(),
|
||||
render_path: state
|
||||
.grammar
|
||||
.as_ref()
|
||||
.and_then(|grammar| grammar.verb(&entry.kind.0))
|
||||
.map(|verb| verb.render_path),
|
||||
roles,
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
|
|
@ -11533,7 +11622,7 @@ mod soul_knowledge_tests {
|
|||
// Nobody holds it.
|
||||
assert!(kernel.state.knowledge.is_empty());
|
||||
// The operator reads it anyway, with the speaker resolved by label.
|
||||
let log = operator_log(&kernel.state).unwrap();
|
||||
let log = kernel.operator_log().unwrap();
|
||||
assert_eq!(
|
||||
log,
|
||||
vec![OperatorEvent {
|
||||
|
|
@ -11542,6 +11631,9 @@ mod soul_knowledge_tests {
|
|||
speaker_label: "The Yard Bystander".into(),
|
||||
speech: Some(Statement::new(text).unwrap()),
|
||||
display: None,
|
||||
verb: "speak".into(),
|
||||
render_path: None,
|
||||
roles: Vec::new(),
|
||||
}]
|
||||
);
|
||||
// And no subject snapshot carries it.
|
||||
|
|
|
|||
|
|
@ -82,6 +82,16 @@ enum Request {
|
|||
AgencyGraph {
|
||||
reply: oneshot::Sender<Result<AgencyGraph, KernelError>>,
|
||||
},
|
||||
/// The tick driver's opportunity list. Separate from `Snapshot` so the
|
||||
/// list a cover is derived from has one door, held by the driver.
|
||||
Opportunities {
|
||||
reply: oneshot::Sender<Result<Vec<DecisionOpportunity>, KernelError>>,
|
||||
},
|
||||
/// The tick ordinal's one door: how many clock advances the world has
|
||||
/// committed. Held by the tick driver, as `Opportunities` is.
|
||||
ClockAdvances {
|
||||
reply: oneshot::Sender<Result<u64, KernelError>>,
|
||||
},
|
||||
ControllerReceipt {
|
||||
command_id: CommandId,
|
||||
opportunity: super::DecisionOpportunity,
|
||||
|
|
@ -486,6 +496,37 @@ impl WorldMailbox {
|
|||
.map_err(MailboxError::Kernel)
|
||||
}
|
||||
|
||||
/// The tick driver's opportunity list, in snapshot order. It is on
|
||||
/// `WorldMailbox` and on no controller port, as `agency_graph` is.
|
||||
pub async fn opportunities(&self) -> Result<Vec<DecisionOpportunity>, MailboxError> {
|
||||
let (reply, response) = oneshot::channel();
|
||||
self.sender
|
||||
.send(Request::Opportunities { reply })
|
||||
.await
|
||||
.map_err(|_| MailboxError::Unavailable)?;
|
||||
|
||||
response
|
||||
.await
|
||||
.map_err(|_| MailboxError::Unavailable)?
|
||||
.map_err(MailboxError::Kernel)
|
||||
}
|
||||
|
||||
/// The tick driver's ordinal: how many clock advances the world has
|
||||
/// committed. It is on `WorldMailbox` and on no controller port, as
|
||||
/// `opportunities` is.
|
||||
pub async fn clock_advances(&self) -> Result<u64, MailboxError> {
|
||||
let (reply, response) = oneshot::channel();
|
||||
self.sender
|
||||
.send(Request::ClockAdvances { reply })
|
||||
.await
|
||||
.map_err(|_| MailboxError::Unavailable)?;
|
||||
|
||||
response
|
||||
.await
|
||||
.map_err(|_| MailboxError::Unavailable)?
|
||||
.map_err(MailboxError::Kernel)
|
||||
}
|
||||
|
||||
pub async fn operator_log(&self) -> Result<Vec<OperatorEvent>, MailboxError> {
|
||||
let (reply, response) = oneshot::channel();
|
||||
self.sender
|
||||
|
|
@ -955,6 +996,32 @@ async fn run_owner(mut owned: OwnedWorld, mut receiver: mpsc::Receiver<Request>)
|
|||
return;
|
||||
}
|
||||
}
|
||||
Request::Opportunities { reply } => {
|
||||
let result = match &owned {
|
||||
OwnedWorld::Empty(_) => Err(KernelError::WorldNotCreated),
|
||||
OwnedWorld::Live(kernel) => kernel.opportunities(),
|
||||
};
|
||||
let owner_must_stop = result
|
||||
.as_ref()
|
||||
.is_err_and(KernelError::requires_owner_restart);
|
||||
let _ = reply.send(result);
|
||||
if owner_must_stop {
|
||||
return;
|
||||
}
|
||||
}
|
||||
Request::ClockAdvances { reply } => {
|
||||
let result = match &owned {
|
||||
OwnedWorld::Empty(_) => Err(KernelError::WorldNotCreated),
|
||||
OwnedWorld::Live(kernel) => kernel.clock_advances(),
|
||||
};
|
||||
let owner_must_stop = result
|
||||
.as_ref()
|
||||
.is_err_and(KernelError::requires_owner_restart);
|
||||
let _ = reply.send(result);
|
||||
if owner_must_stop {
|
||||
return;
|
||||
}
|
||||
}
|
||||
Request::ControllerReceipt {
|
||||
command_id,
|
||||
opportunity,
|
||||
|
|
|
|||
|
|
@ -256,7 +256,7 @@ enum DecodedRow {
|
|||
Ledger(RunLedger),
|
||||
}
|
||||
|
||||
fn canonical<T: Serialize + for<'de> Deserialize<'de>>(
|
||||
pub(super) fn canonical<T: Serialize + for<'de> Deserialize<'de>>(
|
||||
row: &CultCacheEnvelope,
|
||||
) -> Result<T, CallRecordError> {
|
||||
let value: T = rmp_serde::from_slice(&row.payload).map_err(|_| CallRecordError::Corrupt)?;
|
||||
|
|
@ -585,6 +585,16 @@ impl RecordingInferencePort {
|
|||
})
|
||||
}
|
||||
|
||||
/// The run this port records.
|
||||
pub fn run_id(&self) -> &RunId {
|
||||
&self.run
|
||||
}
|
||||
|
||||
/// The token cap the run was opened under.
|
||||
pub fn cap(&self) -> u64 {
|
||||
self.cap
|
||||
}
|
||||
|
||||
/// Tokens reserved or charged to the run so far.
|
||||
pub fn spent(&self) -> u64 {
|
||||
self.store.spent(&self.run).unwrap_or(u64::MAX)
|
||||
|
|
|
|||
|
|
@ -1,13 +1,19 @@
|
|||
//! 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, and the
|
||||
//! genesis of the canonical verse world.
|
||||
//! The Aetheria verse run's own organs. The verse binary sits 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, the genesis of the
|
||||
//! canonical verse world, and the manual run that advances it with its record.
|
||||
|
||||
mod calls;
|
||||
mod genesis;
|
||||
mod run;
|
||||
|
||||
pub use genesis::{CreateVerseWorldIntent, VerseGenesisError};
|
||||
pub use calls::{
|
||||
CALL_RECORD_SCHEMA, CallRecordError, CallRecordStore, ProviderCallRecord,
|
||||
RecordingInferencePort, ReplayInferencePort, RunField, RunId, RunStart,
|
||||
};
|
||||
pub use run::{
|
||||
RUN_RECORD_SCHEMA, RecordedCoverBudget, Refusal, RefusalReason, RunClaim, RunOutcome, RunStatus,
|
||||
TickCheckpoint, TickReport, UnattributedCall, VerseRunConfig, VerseRunError, VerseRunRecord, VerseRunStore,
|
||||
call_run_id, run_manual,
|
||||
};
|
||||
|
|
|
|||
1162
crates/ghostlight/src/verse/run.rs
Normal file
1162
crates/ghostlight/src/verse/run.rs
Normal file
File diff suppressed because it is too large
Load diff
2758
crates/ghostlight/src/verse/run/tests.rs
Normal file
2758
crates/ghostlight/src/verse/run/tests.rs
Normal file
File diff suppressed because it is too large
Load diff
Loading…
Reference in a new issue