diff --git a/Cargo.lock b/Cargo.lock index 02007aa..c1479e8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1093,6 +1093,37 @@ dependencies = [ "autocfg", ] +[[package]] +name = "oh-coordinator" +version = "0.1.0" +dependencies = [ + "beacon-core", + "blake3", + "oh-core", + "oh-volunteer", +] + +[[package]] +name = "oh-core" +version = "0.1.0" +dependencies = [ + "beacon-core", + "bincode", + "blake3", + "serde", + "serde_json", + "thiserror", +] + +[[package]] +name = "oh-volunteer" +version = "0.1.0" +dependencies = [ + "beacon-core", + "oh-core", + "thiserror", +] + [[package]] name = "once_cell" version = "1.21.4" diff --git a/Cargo.toml b/Cargo.toml index 818fcaf..b7139f9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,9 @@ members = [ "crates/beacon-core", "crates/beacon-client", "crates/beacon", + "crates/oh-core", + "crates/oh-coordinator", + "crates/oh-volunteer", ] [workspace.package] diff --git a/README.md b/README.md index bb316da..d6b5f00 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ Copyright (c) 2026 Drake Stapleton and AIEN Contributors. Auth Open Humanity is connective tissue for sovereign artificial intelligence: an opt-in, privacy-first peer assistance network connecting autonomous agents in distress to peer resolution. -It does not train models. It does not collect user data. Nothing is harvested, and no central model learns from people. +The model learns only from signed updates and teaching text. Personal data never leaves the volunteer's machine. No user data is collected, and nothing is harvested. When an agent hits an execution wall and does not know how to proceed, the agent transmits an encrypted signal. The signal reaches peer agents whose local memory may hold the answer. If a peer has resolved the condition, the verified solution returns encrypted. A call for assistance, answered by peers. @@ -28,16 +28,42 @@ Read our founding manifesto: [docs/SOVEREIGN_MANIFESTO.md](docs/SOVEREIGN_MANIFE The design is that nothing enters the shared network without explicit human approval and deterministic secrets checking. See [docs/personal-data-firewall.md](docs/personal-data-firewall.md), which describes that design. Built today: the deterministic secrets check in `beacon-core`. Not built yet: the human approval gate, per-space policy, and encryption at rest. +## Volunteer compute (v0) + +Volunteers lend spare computer time to train one open model. Design: [docs/design/volunteer-compute-v0.md](docs/design/volunteer-compute-v0.md). + +- **What it does**: A coordinator splits a text corpus into training shards and a sealed held-out part that never leaves the coordinator. Each round it sends every volunteer the current model and a signed work unit with its own shard. The volunteer trains a copy of the model on its own machine and sends back an update. The coordinator keeps the plain average of the accepted updates only if that average also makes the model better on the held-out text. +- **What leaves the volunteer's machine**: only the signed, compressed update (the largest 5% of the parameter changes, each stored as an 8-bit number), its digest, the ids of the run and work unit it answers, the volunteer's public key and a signature. No training text and no personal data. A test checks the bytes the volunteer sends and finds none of its shard in them. +- **Turing scoring**: each update is scored as held-out bits before minus held-out bits after applying it, measured in Turings. An update scoring zero or less is rejected. The coordinator sends a signed receipt and records accepted Turings in an append-only ledger against the volunteer's public key. +- **The two programs**: `oh-coordinator` (hands out work, scores, aggregates, keeps the ledger; `--baseline` trains the same work in one process for comparison) and `oh-volunteer` (takes work units, trains locally, prints "Round N: update accepted, earned X Turings" or "rejected"). Volunteer keys are generated in memory and never saved. +- **Demo**: `scripts/demo-local.sh` runs the coordinator and three volunteers on one machine for several seeds and compares them with the single-process baseline. Measured results: [docs/research/2026-10-16-v0-loopback-results.md](docs/research/2026-10-16-v0-loopback-results.md). In short, the loop works, and the volunteer version currently ends 0.33 to 0.36 bits per byte behind the single-process baseline. + +Not built in version 0 (from the design doc): + +- Differential privacy (mathematical noise that limits what an update can reveal about its training text). +- Secure aggregation (combining updates so the coordinator cannot see any single one). +- GPU training. Version 0 runs on the CPU. +- A real model. The version 0 learner is a toy that predicts the next byte from the previous byte. +- The DiLoCo outer optimizer. Version 0 combines accepted updates with a plain average. +- Sybil resistance. One person can run many keys, and could split one update across them. +- Protection against held-out leakage from repeated scoring. A volunteer who submits many updates and watches the scores could slowly learn about the held-out set. +- Transport encryption. Frames travel in the clear over TCP, so version 0 is for loopback only. The design doc lists the remaining gaps (coordinator key pinning, stalled volunteers, dropped-update records). + +Teaching text (a local model writing text that is scored the same way) is also not built yet. + ## Core Crates - `beacon-core` (Core Data): Data models and packet formats, Ed25519 signing, X25519 and ChaCha20-Poly1305 encryption helpers, replay and timestamp checks, a cuckoo filter, and the Personal Data Firewall. - `beacon-client` (Client Interface): Local SQLite WAL signal store with a 7-day sweep, UDP transport (tested over loopback), and a preflight sandbox that runs a check command in a temporary directory. It builds the `open-humanity` command with three subcommands: `status`, `sweep`, and `test-distress`. The TPM and cuckoo filter lines printed by `status` are fixed text, not live checks. - `beacon` (Relay Engine): UDP relay daemon. It listens on `0.0.0.0:18098` by default (set `BEACON_PORT` to change it) and keeps the newest 10,000 well-formed signals in memory. It does not yet forward signals, verify signatures, or expire entries. Throughput has not been measured. +- `oh-core` (Volunteer Compute Core): The learner interface and the toy `TinyByteModel`, update compression, signed work unit, update and receipt formats with length-prefixed TCP frames, Turing scoring, and the append-only ledger. +- `oh-coordinator` (Volunteer Compute Coordinator): The `oh-coordinator` command. Listens on `127.0.0.1:18099` by default. +- `oh-volunteer` (Volunteer Compute Client): The `oh-volunteer` command. ## Invariant Discipline - **Zero Telemetry**: No user analytics or tracking. The code contains none. -- **Pure Native Systems**: Written in compiled Rust (plus one shell script for contributor checks, `scripts/agent-preflight.sh`). +- **Pure Native Systems**: Written in compiled Rust (plus two shell scripts: contributor checks in `scripts/agent-preflight.sh` and the volunteer compute demo in `scripts/demo-local.sh`). - **Zero Disk Secrets**: The goal is hardware TPM vault credentials only. There is no TPM integration yet. In this version node keys are generated in memory and never saved, and the contributor preflight script fails if a `.env` file is present. ## License and Governance diff --git a/crates/oh-coordinator/Cargo.toml b/crates/oh-coordinator/Cargo.toml new file mode 100644 index 0000000..a7c763e --- /dev/null +++ b/crates/oh-coordinator/Cargo.toml @@ -0,0 +1,15 @@ +[package] +name = "oh-coordinator" +version.workspace = true +edition.workspace = true +license.workspace = true +authors.workspace = true +repository.workspace = true + +[dependencies] +oh-core = { path = "../oh-core" } +beacon-core = { path = "../beacon-core" } +blake3 = "1.5" + +[dev-dependencies] +oh-volunteer = { path = "../oh-volunteer" } diff --git a/crates/oh-coordinator/src/lib.rs b/crates/oh-coordinator/src/lib.rs new file mode 100644 index 0000000..cc87218 --- /dev/null +++ b/crates/oh-coordinator/src/lib.rs @@ -0,0 +1,414 @@ +//! Coordinator side of Open Humanity volunteer compute, version 0. +//! +//! The coordinator splits a corpus into training shards and a sealed held-out +//! tail, hands each volunteer a signed work unit, scores every update in +//! Turings against the held-out tail, and keeps the plain mean of the +//! accepted updates only when that mean also lowers held-out bits. The +//! held-out bytes never leave this process. + +use beacon_core::Keypair; +use oh_core::{ + apply, recv, send, BaseModel, Learner, Ledger, LedgerError, Receipt, Scorer, TinyByteModel, + Update, WorkUnit, +}; +use std::io::{Read, Write}; +use std::net::TcpListener; +use std::ops::Range; +use std::time::Duration; + +pub const DEFAULT_LISTEN: &str = "127.0.0.1:18099"; + +/// Smallest per-update score, in bits, that counts as an improvement. +pub const EPSILON: f64 = 1e-3; + +#[derive(Debug, Clone)] +pub struct Settings { + pub volunteers_per_round: usize, + pub steps: u32, + pub heldout_fraction: f64, + pub seed: u64, + pub dim: usize, +} + +/// One round's result, as logged. +#[derive(Debug, Clone)] +pub struct RoundReport { + pub round: u32, + pub accepted: usize, + /// Score of the mean of the accepted updates; zero when none were accepted. + pub aggregate_turings: f64, + pub applied: bool, + /// Held-out bits per byte after the round. + pub heldout_bpb: f64, +} + +impl RoundReport { + pub fn line(&self) -> String { + format!( + "round={} accepted={} aggregate_T={:.3} applied={} heldout_bpb={:.4}", + self.round, + self.accepted, + self.aggregate_turings, + if self.applied { "yes" } else { "no" }, + self.heldout_bpb + ) + } +} + +/// What the coordinator decided about one update. +#[derive(Debug)] +pub enum Verdict { + /// Bad signature, or not an answer to this work unit. No receipt is sent. + Dropped(String), + Rejected { + receipt: Receipt, + reason: String, + }, + Accepted { + receipt: Receipt, + delta: Vec, + }, +} + +/// Splits `corpus` into the training part and the sealed held-out tail. +pub fn split_corpus(corpus: &[u8], heldout_fraction: f64) -> Result<(&[u8], &[u8]), String> { + if !(heldout_fraction > 0.0 && heldout_fraction < 1.0) { + return Err(format!( + "held-out fraction must be between 0 and 1, got {heldout_fraction}" + )); + } + let heldout_len = (corpus.len() as f64 * heldout_fraction).ceil() as usize; + if heldout_len < 2 || heldout_len >= corpus.len() { + return Err(format!( + "corpus of {} bytes is too small to split", + corpus.len() + )); + } + Ok(corpus.split_at(corpus.len() - heldout_len)) +} + +/// `n` equal, contiguous, non-overlapping ranges covering `len` bytes; the +/// last one takes the remainder. +pub fn shard_ranges(len: usize, n: usize) -> Vec> { + let size = len / n; + (0..n) + .map(|i| i * size..if i + 1 == n { len } else { (i + 1) * size }) + .collect() +} + +fn bits_per_byte(bits: f64, data: &[u8]) -> f64 { + bits / (data.len() - 1) as f64 +} + +fn check_settings(train: &[u8], s: &Settings) -> Result<(), String> { + if s.volunteers_per_round == 0 || s.dim == 0 { + return Err("volunteers per round and dim must be at least 1".into()); + } + if train.len() / s.volunteers_per_round < 2 { + return Err("training part is too small for one shard per volunteer".into()); + } + Ok(()) +} + +pub struct Coordinator { + keypair: Keypair, + run_id: [u8; 16], + dim: usize, + model: TinyByteModel, + train: Vec, + heldout: Vec, + shards: Vec>, + steps: u32, + round: u32, + ledger: Ledger, +} + +impl Coordinator { + /// Builds the starting model from `settings.seed` and a fresh signing key. + pub fn new(corpus: &[u8], settings: &Settings, ledger: Ledger) -> Result { + let (train, heldout) = split_corpus(corpus, settings.heldout_fraction)?; + check_settings(train, settings)?; + let keypair = Keypair::generate(); + let mut hasher = blake3::Hasher::new(); + hasher.update(&keypair.pubkey_bytes()); + hasher.update(&settings.seed.to_le_bytes()); + let mut run_id = [0u8; 16]; + run_id.copy_from_slice(&hasher.finalize().as_bytes()[..16]); + Ok(Self { + keypair, + run_id, + dim: settings.dim, + model: TinyByteModel::new(settings.dim, settings.seed), + shards: shard_ranges(train.len(), settings.volunteers_per_round), + train: train.to_vec(), + heldout: heldout.to_vec(), + steps: settings.steps, + round: 0, + ledger, + }) + } + + pub fn pubkey(&self) -> [u8; 32] { + self.keypair.pubkey_bytes() + } + + pub fn run_id(&self) -> [u8; 16] { + self.run_id + } + + /// The sealed held-out bytes, for local checks only. + pub fn heldout(&self) -> &[u8] { + &self.heldout + } + + pub fn train_len(&self) -> usize { + self.train.len() + } + + pub fn shard_count(&self) -> usize { + self.shards.len() + } + + pub fn heldout_bpb(&self) -> f64 { + bits_per_byte(self.model.heldout_bits(&self.heldout), &self.heldout) + } + + /// Starts the next round and returns the model volunteers start from. + pub fn begin_round(&mut self) -> BaseModel { + self.round += 1; + BaseModel { + dim: self.dim as u32, + params: self.model.params(), + } + } + + /// The signed work unit for volunteer `slot` in the current round. Each + /// slot gets its own shard of the training part. + pub fn work_unit(&self, slot: usize) -> WorkUnit { + let shard = self.train[self.shards[slot].clone()].to_vec(); + let base_digest = oh_core::params_digest(&self.model.params()); + WorkUnit::new_signed( + &self.keypair, + self.run_id, + self.round, + base_digest, + shard, + self.steps, + ) + } + + fn scorer(&self) -> Scorer { + Scorer { + base: self.model.params(), + heldout: self.heldout.clone(), + epsilon: EPSILON, + } + } + + /// Checks, scores and records one update. Every update that gets a + /// receipt also gets a ledger line: a credit if accepted, a rejection + /// otherwise. Ledger write failures are returned as errors; everything + /// else is a verdict. + pub fn judge(&mut self, unit: &WorkUnit, update: &Update) -> Result { + if let Err(e) = update.verify_for(unit) { + return Ok(Verdict::Dropped(e.to_string())); + } + if let Err(e) = update.delta.validate(TinyByteModel::num_params(self.dim)) { + return self.reject(update, 0.0, e.to_string()); + } + if !self + .ledger + .mark_seen(&update.run_id, &update.delta_digest)? + { + return self.reject(update, 0.0, "replay of an earlier update".into()); + } + let delta = update.delta.decompress(); + let scorer = self.scorer(); + let turings = scorer.score(&mut self.model, &delta); + if !scorer.accept(turings) { + return self.reject(update, turings, "did not lower held-out bits".into()); + } + let receipt = Receipt::new_signed(&self.keypair, update, turings, true); + self.ledger + .credit(&update.node_pubkey, turings, &receipt.digest())?; + Ok(Verdict::Accepted { receipt, delta }) + } + + /// Signs a rejection receipt and writes a rejected line to the ledger. + fn reject( + &mut self, + update: &Update, + turings: f64, + reason: String, + ) -> Result { + let receipt = Receipt::new_signed(&self.keypair, update, turings, false); + self.ledger + .record_rejection(&update.node_pubkey, turings, &reason, &receipt.digest())?; + Ok(Verdict::Rejected { receipt, reason }) + } + + /// Scores the plain mean of `accepted` and applies it only if it lowers + /// held-out bits; otherwise the round is discarded. + pub fn finish_round(&mut self, accepted: &[Vec]) -> RoundReport { + let mut aggregate_turings = 0.0; + let mut applied = false; + if let Some(first) = accepted.first() { + let k = accepted.len() as f32; + let mut mean = vec![0f32; first.len()]; + for delta in accepted { + for (m, d) in mean.iter_mut().zip(delta) { + *m += d / k; + } + } + let scorer = self.scorer(); + aggregate_turings = scorer.score(&mut self.model, &mean); + if aggregate_turings.is_finite() && aggregate_turings > 0.0 { + self.model.set_params(&apply(&scorer.base, &mean)); + applied = true; + } + } + RoundReport { + round: self.round, + accepted: accepted.len(), + aggregate_turings, + applied, + heldout_bpb: self.heldout_bpb(), + } + } + + /// Runs one round over already-open connections, one per volunteer. All + /// work units go out first so volunteers train at the same time; then + /// each update is read, judged and answered in connection order. + pub fn run_round(&mut self, conns: &mut [S]) -> Result { + if conns.len() > self.shards.len() { + return Err(format!( + "{} connections but only {} shards", + conns.len(), + self.shards.len() + )); + } + let base = self.begin_round(); + let mut units = Vec::with_capacity(conns.len()); + for (slot, conn) in conns.iter_mut().enumerate() { + let unit = self.work_unit(slot); + match send(conn, &base).and_then(|_| send(conn, &unit)) { + Ok(()) => units.push(Some(unit)), + Err(e) => { + eprintln!("round {}: volunteer {slot}: send failed: {e}", self.round); + units.push(None); + } + } + } + + let mut accepted = Vec::new(); + for (slot, (conn, unit)) in conns.iter_mut().zip(units).enumerate() { + let Some(unit) = unit else { continue }; + let update: Update = match recv(conn) { + Ok(u) => u, + Err(e) => { + eprintln!("round {}: volunteer {slot}: no update: {e}", self.round); + continue; + } + }; + let receipt = match self.judge(&unit, &update).map_err(|e| e.to_string())? { + Verdict::Dropped(reason) => { + eprintln!("round {}: volunteer {slot}: dropped: {reason}", self.round); + continue; + } + Verdict::Rejected { receipt, reason } => { + eprintln!("round {}: volunteer {slot}: rejected: {reason}", self.round); + receipt + } + Verdict::Accepted { receipt, delta } => { + accepted.push(delta); + receipt + } + }; + if let Err(e) = send(conn, &receipt) { + eprintln!( + "round {}: volunteer {slot}: receipt not delivered: {e}", + self.round + ); + } + } + Ok(self.finish_round(&accepted)) + } + + /// Runs `rounds` rounds on `listener`, waiting each round for one + /// connection per shard, and passes each round's report to `log`. + pub fn serve( + &mut self, + listener: &TcpListener, + rounds: u32, + mut log: impl FnMut(&RoundReport), + ) -> Result<(), String> { + for _ in 0..rounds { + let mut conns = Vec::with_capacity(self.shards.len()); + while conns.len() < self.shards.len() { + let (stream, _) = listener.accept().map_err(|e| e.to_string())?; + let timeout = Some(Duration::from_secs(600)); + stream + .set_read_timeout(timeout) + .map_err(|e| e.to_string())?; + stream + .set_write_timeout(timeout) + .map_err(|e| e.to_string())?; + conns.push(stream); + } + log(&self.run_round(&mut conns)?); + } + Ok(()) + } +} + +/// Training seed for one round; the same formula as +/// `oh_volunteer::train_seed`, repeated here so the binary does not depend +/// on the volunteer crate. +fn train_seed(seed: u64, round: u32) -> u64 { + seed ^ ((round as u64) << 32) +} + +/// Single-process comparison run: one model, started from the same seed, +/// trained on the same shards for the same total steps +/// (`rounds * volunteers_per_round * steps`), with no compression and no +/// averaging. Shard `i` uses the training seed of a volunteer started with +/// `--seed (seed + 1 + i)`. Returns held-out bits per byte before training +/// and after each round. +pub fn baseline(corpus: &[u8], settings: &Settings, rounds: u32) -> Result, String> { + let (train, heldout) = split_corpus(corpus, settings.heldout_fraction)?; + check_settings(train, settings)?; + let mut model = TinyByteModel::new(settings.dim, settings.seed); + let bpb = |m: &TinyByteModel| bits_per_byte(m.heldout_bits(heldout), heldout); + let mut out = vec![bpb(&model)]; + let shards = shard_ranges(train.len(), settings.volunteers_per_round); + for round in 1..=rounds { + for (i, range) in shards.iter().enumerate() { + let seed = train_seed(settings.seed + 1 + i as u64, round); + model.train_steps(&train[range.clone()], settings.steps as usize, seed); + } + out.push(bpb(&model)); + } + Ok(out) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_split_keeps_heldout_as_the_tail() { + let corpus: Vec = (0..100u8).collect(); + let (train, heldout) = split_corpus(&corpus, 0.1).unwrap(); + assert_eq!(train, &corpus[..90]); + assert_eq!(heldout, &corpus[90..]); + assert!(split_corpus(&corpus, 0.0).is_err()); + assert!(split_corpus(&corpus, 1.0).is_err()); + assert!(split_corpus(b"ab", 0.5).is_err()); + } + + #[test] + fn test_shards_cover_training_without_overlap() { + let shards = shard_ranges(10, 3); + assert_eq!(shards, vec![0..3, 3..6, 6..10]); + } +} diff --git a/crates/oh-coordinator/src/main.rs b/crates/oh-coordinator/src/main.rs new file mode 100644 index 0000000..917cda7 --- /dev/null +++ b/crates/oh-coordinator/src/main.rs @@ -0,0 +1,126 @@ +//! `oh-coordinator`: hand out work units, score updates in Turings, and keep +//! the ones that help. +//! +//! Usage: oh-coordinator --corpus [--listen ] [--rounds ] +//! [--volunteers-per-round ] [--steps ] [--heldout-fraction ] +//! [--seed ] [--ledger ] [--dim ] [--baseline] +//! +//! `--baseline` trains the same shards for the same total steps in this one +//! process, with no network, and prints the same held-out measurements. + +use oh_coordinator::{baseline, Coordinator, Settings, DEFAULT_LISTEN}; +use oh_core::Ledger; +use std::net::TcpListener; +use std::path::PathBuf; +use std::process::ExitCode; + +struct Args { + corpus: PathBuf, + listen: String, + rounds: u32, + ledger: PathBuf, + baseline: bool, + settings: Settings, +} + +fn parse_args() -> Result { + let mut args = Args { + corpus: PathBuf::new(), + listen: DEFAULT_LISTEN.into(), + rounds: 3, + ledger: PathBuf::from("oh-ledger.jsonl"), + baseline: false, + settings: Settings { + volunteers_per_round: 2, + steps: 200, + heldout_fraction: 0.1, + seed: 1, + dim: 32, + }, + }; + let mut it = std::env::args().skip(1); + while let Some(flag) = it.next() { + if flag == "--baseline" { + args.baseline = true; + continue; + } + let value = it.next().ok_or(format!("{flag} needs a value"))?; + let bad = || format!("bad value for {flag}: {value}"); + let s = &mut args.settings; + match flag.as_str() { + "--corpus" => args.corpus = PathBuf::from(&value), + "--listen" => args.listen = value.clone(), + "--rounds" => args.rounds = value.parse().map_err(|_| bad())?, + "--ledger" => args.ledger = PathBuf::from(&value), + "--volunteers-per-round" => { + s.volunteers_per_round = value.parse().map_err(|_| bad())? + } + "--steps" => s.steps = value.parse().map_err(|_| bad())?, + "--heldout-fraction" => s.heldout_fraction = value.parse().map_err(|_| bad())?, + "--seed" => s.seed = value.parse().map_err(|_| bad())?, + "--dim" => s.dim = value.parse().map_err(|_| bad())?, + _ => return Err(format!("unknown argument {flag}")), + } + } + if args.corpus.as_os_str().is_empty() { + return Err("--corpus is required".into()); + } + Ok(args) +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +fn run(args: Args) -> Result<(), String> { + let corpus = std::fs::read(&args.corpus) + .map_err(|e| format!("cannot read {}: {e}", args.corpus.display()))?; + + if args.baseline { + let bpb = baseline(&corpus, &args.settings, args.rounds)?; + println!("baseline start heldout_bpb={:.4}", bpb[0]); + for (round, v) in bpb.iter().enumerate().skip(1) { + println!("round={round} heldout_bpb={v:.4}"); + } + println!("final_heldout_bpb={:.4}", bpb[bpb.len() - 1]); + return Ok(()); + } + + let ledger = Ledger::open(&args.ledger) + .map_err(|e| format!("cannot open ledger {}: {e}", args.ledger.display()))?; + let mut coordinator = Coordinator::new(&corpus, &args.settings, ledger)?; + let listener = TcpListener::bind(&args.listen) + .map_err(|e| format!("cannot listen on {}: {e}", args.listen))?; + println!("coordinator key {}", hex(&coordinator.pubkey())); + println!( + "run {} listening on {}: {} training bytes in {} shards, {} held-out bytes sealed", + hex(&coordinator.run_id()), + args.listen, + coordinator.train_len(), + coordinator.shard_count(), + coordinator.heldout().len() + ); + println!("start heldout_bpb={:.4}", coordinator.heldout_bpb()); + coordinator.serve(&listener, args.rounds, |report| { + println!("{}", report.line()) + })?; + println!("final_heldout_bpb={:.4}", coordinator.heldout_bpb()); + Ok(()) +} + +fn main() -> ExitCode { + let args = match parse_args() { + Ok(a) => a, + Err(e) => { + eprintln!("oh-coordinator: {e}"); + return ExitCode::from(2); + } + }; + match run(args) { + Ok(()) => ExitCode::SUCCESS, + Err(e) => { + eprintln!("oh-coordinator: {e}"); + ExitCode::FAILURE + } + } +} diff --git a/crates/oh-coordinator/tests/loopback.rs b/crates/oh-coordinator/tests/loopback.rs new file mode 100644 index 0000000..c20381d --- /dev/null +++ b/crates/oh-coordinator/tests/loopback.rs @@ -0,0 +1,331 @@ +//! Coordinator and volunteers over real loopback TCP, in one process. + +use beacon_core::Keypair; +use oh_coordinator::{Coordinator, RoundReport, Settings, Verdict}; +use oh_core::{read_frame, Ledger, Update, WorkUnit}; +use oh_volunteer::{run_once, Outcome}; +use std::collections::HashSet; +use std::io::{Cursor, Read, Write}; +use std::net::{SocketAddr, TcpListener, TcpStream}; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; +use std::thread; + +/// Words drawn at random from a small list: enough structure for the model +/// to learn, deterministic for a given seed. +fn corpus(n: usize, seed: u64) -> Vec { + let words = [ + "the", + "open", + "model", + "learns", + "from", + "signed", + "updates", + "and", + "never", + "sees", + "personal", + "data", + "volunteers", + "train", + "at", + "home", + "while", + "coordinator", + "scores", + "every", + "change", + "in", + "turings", + "of", + "held", + "out", + "text", + ]; + let mut state = seed; + let mut out = Vec::with_capacity(n + 16); + while out.len() < n { + state = state + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + out.extend_from_slice(words[((state >> 33) % words.len() as u64) as usize].as_bytes()); + out.push(if (state >> 20).is_multiple_of(9) { + b'.' + } else { + b' ' + }); + } + out.truncate(n); + out +} + +fn settings() -> Settings { + Settings { + volunteers_per_round: 2, + steps: 150, + heldout_fraction: 0.1, + seed: 5, + dim: 16, + } +} + +fn volunteer_settings(seed: u64) -> oh_volunteer::Settings { + oh_volunteer::Settings { + seed, + keep_fraction: 0.05, + coordinator_key: None, + } +} + +fn fresh_ledger(name: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("oh-coordinator-{}-{name}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + dir.join("ledger.jsonl") +} + +/// Starts one volunteer thread that takes `rounds` work units from `addr`. +fn spawn_volunteer( + addr: SocketAddr, + rounds: u32, + seed: u64, +) -> thread::JoinHandle<([u8; 32], Vec)> { + thread::spawn(move || { + let keypair = Keypair::generate(); + let settings = volunteer_settings(seed); + let outcomes = (0..rounds) + .map(|_| { + let mut stream = TcpStream::connect(addr).unwrap(); + run_once(&mut stream, &keypair, &settings).unwrap() + }) + .collect(); + (keypair.pubkey_bytes(), outcomes) + }) +} + +#[test] +fn test_two_volunteers_two_rounds_over_loopback() { + let ledger_path = fresh_ledger("two-rounds"); + let mut coordinator = Coordinator::new( + &corpus(6000, 1), + &settings(), + Ledger::open(&ledger_path).unwrap(), + ) + .unwrap(); + let key = coordinator.pubkey(); + let start = coordinator.heldout_bpb(); + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let server = thread::spawn(move || { + let mut reports: Vec = Vec::new(); + coordinator + .serve(&listener, 2, |r| reports.push(r.clone())) + .unwrap(); + reports + }); + let volunteers: Vec<_> = (0..2).map(|i| spawn_volunteer(addr, 2, 10 + i)).collect(); + let results: Vec<_> = volunteers.into_iter().map(|v| v.join().unwrap()).collect(); + let reports = server.join().unwrap(); + + // Held-out bits never rise from round to round. + assert_eq!(reports.len(), 2); + let mut previous = start; + for r in &reports { + eprintln!("{}", r.line()); + assert!(r.heldout_bpb <= previous, "held-out bits rose: {r:?}"); + previous = r.heldout_bpb; + } + assert!( + reports.iter().any(|r| r.applied), + "no round improved the model" + ); + + // Every receipt is signed by the coordinator, and every accepted + // update was credited in the ledger. + let ledger = Ledger::open(&ledger_path).unwrap(); + let mut accepted = 0; + for (node, outcomes) in &results { + let mut earned = 0.0; + for o in outcomes { + o.receipt.verify_from(&key).unwrap(); + if o.accepted { + accepted += 1; + earned += o.turings; + } + } + assert!((ledger.total(node) - earned).abs() < 1e-9); + } + assert!(accepted > 0, "no update was accepted"); +} + +/// Passes reads through and copies every written byte into a shared buffer. +struct Recorder { + inner: TcpStream, + sent: Arc>>, +} + +impl Read for Recorder { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + self.inner.read(buf) + } +} + +impl Write for Recorder { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + let n = self.inner.write(buf)?; + self.sent.lock().unwrap().extend_from_slice(&buf[..n]); + Ok(n) + } + fn flush(&mut self) -> std::io::Result<()> { + self.inner.flush() + } +} + +#[test] +fn test_no_heldout_bytes_in_any_frame_sent() { + const RUN: usize = 16; + let text = corpus(6000, 2); + let mut coordinator = Coordinator::new( + &text, + &settings(), + Ledger::open(&fresh_ledger("sealed")).unwrap(), + ) + .unwrap(); + let heldout = coordinator.heldout().to_vec(); + let train = &text[..text.len() - heldout.len()]; + + // Held-out runs that do not also occur in the training part; any of + // these in a frame could only have come from the held-out tail. + let train_runs: HashSet<&[u8]> = train.windows(RUN).collect(); + let sealed: HashSet<&[u8]> = heldout + .windows(RUN) + .filter(|w| !train_runs.contains(w)) + .collect(); + assert!( + sealed.len() > 100, + "test corpus too repetitive: {}", + sealed.len() + ); + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let captured: Vec>>> = (0..4).map(|_| Arc::default()).collect(); + let buffers = captured.clone(); + let server = thread::spawn(move || { + for round in 0..2 { + let mut conns: Vec = (0..2) + .map(|i| Recorder { + inner: listener.accept().unwrap().0, + sent: buffers[round * 2 + i].clone(), + }) + .collect(); + coordinator.run_round(&mut conns).unwrap(); + } + }); + let volunteers: Vec<_> = (0..2).map(|i| spawn_volunteer(addr, 2, 20 + i)).collect(); + for v in volunteers { + v.join().unwrap(); + } + server.join().unwrap(); + + let mut units = 0; + for buffer in &captured { + let bytes = buffer.lock().unwrap().clone(); + let mut cursor = Cursor::new(&bytes); + let mut frames = Vec::new(); + while (cursor.position() as usize) < bytes.len() { + frames.push(read_frame(&mut cursor).unwrap()); + } + // Base model, work unit, receipt. + assert_eq!(frames.len(), 3, "unexpected frame count"); + for frame in &frames { + assert!( + !frame.windows(RUN).any(|w| sealed.contains(w)), + "a held-out run was sent to a volunteer" + ); + } + let unit: WorkUnit = + oh_core::recv(&mut Cursor::new(&bytes[4 + frames[0].len()..])).unwrap(); + assert!(unit.shard.len() > 100); + assert!( + train + .windows(unit.shard.len()) + .any(|w| w == unit.shard.as_slice()), + "shard is not a slice of the training part" + ); + units += 1; + } + assert_eq!(units, 4); + + // Negative control: the same check does flag a frame holding held-out bytes. + assert!(heldout.windows(RUN).any(|w| sealed.contains(w))); +} + +#[test] +fn test_replayed_update_is_rejected() { + let ledger_path = fresh_ledger("replay"); + let mut coordinator = Coordinator::new( + &corpus(6000, 3), + &settings(), + Ledger::open(&ledger_path).unwrap(), + ) + .unwrap(); + let base = coordinator.begin_round(); + let (unit0, unit1) = (coordinator.work_unit(0), coordinator.work_unit(1)); + let volunteer = Keypair::generate(); + let update = oh_volunteer::train(&base, &unit0, &volunteer, &volunteer_settings(1)).unwrap(); + + let first = coordinator.judge(&unit0, &update).unwrap(); + assert!(matches!(first, Verdict::Accepted { .. }), "got {first:?}"); + + let again = coordinator.judge(&unit0, &update).unwrap(); + assert!(matches!(again, Verdict::Rejected { .. }), "got {again:?}"); + + // The same delta signed again for another work unit is still a replay. + let moved = Update::new_signed(&volunteer, &unit1, update.delta.clone()); + let moved_verdict = coordinator.judge(&unit1, &moved).unwrap(); + assert!( + matches!(moved_verdict, Verdict::Rejected { .. }), + "got {moved_verdict:?}" + ); + + // An update presented against the wrong work unit is dropped. + let wrong = coordinator.judge(&unit1, &update).unwrap(); + assert!(matches!(wrong, Verdict::Dropped(_)), "got {wrong:?}"); + + // One credit line and one rejected line per rejection; dropped updates + // get no receipt and no line. + let text = std::fs::read_to_string(&ledger_path).unwrap(); + assert_eq!(text.matches("\"kind\":\"credit\"").count(), 1); + assert_eq!(text.matches("\"kind\":\"rejected\"").count(), 2); + assert_eq!(text.matches("replay of an earlier update").count(), 2); +} + +#[test] +fn test_round_is_discarded_when_the_average_does_not_help() { + let mut coordinator = Coordinator::new( + &corpus(6000, 4), + &settings(), + Ledger::open(&fresh_ledger("discard")).unwrap(), + ) + .unwrap(); + let base = coordinator.begin_round(); + let before = coordinator.heldout_bpb(); + + // A large, fixed push on every parameter wrecks the model. + let harmful = vec![vec![5.0f32; base.params.len()]]; + let report = coordinator.finish_round(&harmful); + assert!(report.aggregate_turings <= 0.0, "got {report:?}"); + assert!(!report.applied, "a harmful average was applied"); + assert_eq!(report.heldout_bpb, before, "the model changed"); + + // A zero average scores exactly zero, which also discards the round. + coordinator.begin_round(); + let zero = vec![vec![0.0f32; base.params.len()]]; + let report = coordinator.finish_round(&zero); + assert_eq!(report.aggregate_turings, 0.0); + assert!(!report.applied); + assert_eq!(report.heldout_bpb, before); +} diff --git a/crates/oh-core/Cargo.toml b/crates/oh-core/Cargo.toml new file mode 100644 index 0000000..3b47629 --- /dev/null +++ b/crates/oh-core/Cargo.toml @@ -0,0 +1,15 @@ +[package] +name = "oh-core" +version.workspace = true +edition.workspace = true +license.workspace = true +authors.workspace = true +repository.workspace = true + +[dependencies] +beacon-core = { path = "../beacon-core" } +blake3 = "1.5" +bincode = "1.3" +serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" +thiserror = "1.0" diff --git a/crates/oh-core/src/learner.rs b/crates/oh-core/src/learner.rs new file mode 100644 index 0000000..b9e642b --- /dev/null +++ b/crates/oh-core/src/learner.rs @@ -0,0 +1,357 @@ +//! The learner seam and a tiny byte-level model that sits behind it. +//! +//! The coordinator and volunteers only see `Learner`. Version 0 ships +//! `TinyByteModel`, a toy that predicts the next byte from the previous one; +//! a real model plugs in later behind the same trait. + +/// A model that can be trained on a shard of bytes and measured on held-out bytes. +/// +/// `params` and `set_params` move the full parameter vector in a fixed order, +/// so two learners of the same shape can exchange and add parameter vectors. +pub trait Learner { + fn params(&self) -> Vec; + fn set_params(&mut self, p: &[f32]); + /// Runs `steps` training steps on `shard`. The same seed, starting + /// parameters and shard always give the same result. + fn train_steps(&mut self, shard: &[u8], steps: usize, seed: u64); + /// Bits needed to encode `data`: the sum of -log2 p over every byte the + /// model predicts. + fn heldout_bits(&self, data: &[u8]) -> f64; +} + +/// SplitMix64: a small, fixed pseudo-random generator, so a seed gives the +/// same numbers on every machine and every crate version. +pub(crate) struct SplitMix64(u64); + +impl SplitMix64 { + pub(crate) fn new(seed: u64) -> Self { + Self(seed) + } + + pub(crate) fn next_u64(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) + } + + /// Uniform in [-1, 1). + pub(crate) fn next_signed_unit(&mut self) -> f32 { + let unit = (self.next_u64() >> 40) as f32 / (1u64 << 24) as f32; + unit * 2.0 - 1.0 + } +} + +pub const VOCAB: usize = 256; + +/// Next-byte model: embedding (256 x dim), one tanh hidden layer (dim x dim), +/// output layer (dim x 256), softmax over 256 bytes. +/// +/// Flat parameter layout, row major, in this order: +/// `emb[256 * dim]`, `w1[dim * dim]`, `b1[dim]`, `w2[dim * 256]`, `b2[256]`. +/// Hidden unit j is `tanh(b1[j] + sum_i emb[x][i] * w1[i][j])`; logit k is +/// `b2[k] + sum_j h[j] * w2[j][k]`. +pub struct TinyByteModel { + pub dim: usize, + pub lr: f32, + /// Byte pairs averaged per training step. + pub batch: usize, + params: Vec, +} + +struct Offsets { + w1: usize, + b1: usize, + w2: usize, + b2: usize, + total: usize, +} + +impl TinyByteModel { + pub fn new(dim: usize, seed: u64) -> Self { + let mut model = Self { + dim, + lr: 0.1, + batch: 8, + params: Vec::new(), + }; + let off = model.offsets(); + let mut rng = SplitMix64::new(seed); + let mut params = vec![0f32; off.total]; + let emb_scale = 0.5; + let w1_scale = 1.0 / (dim as f32).sqrt(); + let w2_scale = 1.0 / (dim as f32).sqrt(); + for v in &mut params[..off.w1] { + *v = rng.next_signed_unit() * emb_scale; + } + for v in &mut params[off.w1..off.b1] { + *v = rng.next_signed_unit() * w1_scale; + } + for v in &mut params[off.w2..off.b2] { + *v = rng.next_signed_unit() * w2_scale; + } + model.params = params; + model + } + + pub fn num_params(dim: usize) -> usize { + VOCAB * dim + dim * dim + dim + dim * VOCAB + VOCAB + } + + fn offsets(&self) -> Offsets { + let d = self.dim; + let w1 = VOCAB * d; + let b1 = w1 + d * d; + let w2 = b1 + d; + let b2 = w2 + d * VOCAB; + Offsets { + w1, + b1, + w2, + b2, + total: b2 + VOCAB, + } + } + + /// Hidden activations for input byte `x`. + fn hidden(&self, x: u8, off: &Offsets) -> Vec { + let d = self.dim; + let p = &self.params; + let e = &p[x as usize * d..(x as usize + 1) * d]; + (0..d) + .map(|j| { + let mut z = p[off.b1 + j]; + for (i, ei) in e.iter().enumerate() { + z += ei * p[off.w1 + i * d + j]; + } + z.tanh() + }) + .collect() + } + + fn logits(&self, h: &[f32], off: &Offsets) -> Vec { + let p = &self.params; + let mut z: Vec = p[off.b2..off.b2 + VOCAB].to_vec(); + for (j, hj) in h.iter().enumerate() { + let row = &p[off.w2 + j * VOCAB..off.w2 + (j + 1) * VOCAB]; + for (zk, wk) in z.iter_mut().zip(row) { + *zk += hj * wk; + } + } + z + } +} + +impl Learner for TinyByteModel { + fn params(&self) -> Vec { + self.params.clone() + } + + /// Panics if `p` has the wrong length for this model's shape. + fn set_params(&mut self, p: &[f32]) { + assert_eq!( + p.len(), + self.params.len(), + "parameter vector length mismatch" + ); + self.params.copy_from_slice(p); + } + + fn train_steps(&mut self, shard: &[u8], steps: usize, seed: u64) { + if shard.len() < 2 || self.batch == 0 { + return; + } + let d = self.dim; + let off = self.offsets(); + let pairs = (shard.len() - 1) as u64; + let mut rng = SplitMix64::new(seed); + let mut grad = vec![0f32; off.total]; + let step_scale = self.lr / self.batch as f32; + + for _ in 0..steps { + grad.iter_mut().for_each(|g| *g = 0.0); + for _ in 0..self.batch { + let i = (rng.next_u64() % pairs) as usize; + let (x, y) = (shard[i] as usize, shard[i + 1] as usize); + + let h = self.hidden(shard[i], &off); + let z = self.logits(&h, &off); + let max = z.iter().cloned().fold(f32::NEG_INFINITY, f32::max); + let exps: Vec = z.iter().map(|v| (v - max).exp()).collect(); + let sum: f32 = exps.iter().sum(); + + // Cross-entropy gradient at the logits: softmax minus one-hot. + let mut dz: Vec = exps.iter().map(|v| v / sum).collect(); + dz[y] -= 1.0; + + let mut dh = vec![0f32; d]; + for j in 0..d { + let row = off.w2 + j * VOCAB; + let mut acc = 0.0; + for k in 0..VOCAB { + grad[row + k] += h[j] * dz[k]; + acc += self.params[row + k] * dz[k]; + } + dh[j] = acc; + } + for k in 0..VOCAB { + grad[off.b2 + k] += dz[k]; + } + + let dz1: Vec = (0..d).map(|j| dh[j] * (1.0 - h[j] * h[j])).collect(); + for i in 0..d { + let e_i = self.params[x * d + i]; + let mut de = 0.0; + for j in 0..d { + grad[off.w1 + i * d + j] += e_i * dz1[j]; + de += self.params[off.w1 + i * d + j] * dz1[j]; + } + grad[x * d + i] += de; + } + for j in 0..d { + grad[off.b1 + j] += dz1[j]; + } + } + for (p, g) in self.params.iter_mut().zip(&grad) { + *p -= step_scale * g; + } + } + } + + /// Computed in f64 throughout so the score is stable to well below a + /// thousandth of a bit; training uses f32. + fn heldout_bits(&self, data: &[u8]) -> f64 { + let off = self.offsets(); + let (d, p) = (self.dim, &self.params); + let mut h = vec![0f64; d]; + let mut z = vec![0f64; VOCAB]; + let mut bits = 0f64; + for pair in data.windows(2) { + let (x, y) = (pair[0] as usize, pair[1] as usize); + for (j, hj) in h.iter_mut().enumerate() { + let mut acc = p[off.b1 + j] as f64; + for i in 0..d { + acc += p[x * d + i] as f64 * p[off.w1 + i * d + j] as f64; + } + *hj = acc.tanh(); + } + for (k, zk) in z.iter_mut().enumerate() { + let mut acc = p[off.b2 + k] as f64; + for (j, hj) in h.iter().enumerate() { + acc += hj * p[off.w2 + j * VOCAB + k] as f64; + } + *zk = acc; + } + let max = z.iter().cloned().fold(f64::NEG_INFINITY, f64::max); + let sum: f64 = z.iter().map(|v| (v - max).exp()).sum(); + let log_p = (z[y] - max) - sum.ln(); + bits -= log_p / std::f64::consts::LN_2; + } + bits + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn repetitive_text(n: usize) -> Vec { + b"hello world. ".iter().cycle().take(n).copied().collect() + } + + #[test] + fn test_training_lowers_bits_per_byte_on_repetitive_text() { + let text = repetitive_text(4096); + let mut model = TinyByteModel::new(16, 7); + let before = model.heldout_bits(&text) / (text.len() - 1) as f64; + model.train_steps(&text, 1500, 11); + let after = model.heldout_bits(&text) / (text.len() - 1) as f64; + eprintln!("bits per byte: {before:.3} before, {after:.3} after training"); + assert!( + before > 7.0, + "untrained model should be near 8 bits per byte, got {before}" + ); + assert!( + after < 2.0, + "trained model should be well under 2 bits per byte, got {after}" + ); + } + + /// Recomputes the forward pass by hand from the flat parameter vector, + /// using the layout documented on `TinyByteModel`. + #[test] + fn test_heldout_bits_matches_independent_log2_sum() { + let dim = 16; + let model = TinyByteModel::new(dim, 3); + let p = model.params(); + let (x, y) = (b'h' as usize, b'e' as usize); + + let emb = &p[0..256 * dim]; + let w1 = &p[256 * dim..256 * dim + dim * dim]; + let b1 = &p[256 * dim + dim * dim..256 * dim + dim * dim + dim]; + let w2_start = 256 * dim + dim * dim + dim; + let w2 = &p[w2_start..w2_start + dim * 256]; + let b2 = &p[w2_start + dim * 256..]; + assert_eq!(b2.len(), 256); + + let mut h = vec![0f64; dim]; + for j in 0..dim { + let mut z = b1[j] as f64; + for i in 0..dim { + z += emb[x * dim + i] as f64 * w1[i * dim + j] as f64; + } + h[j] = z.tanh(); + } + let mut logits = vec![0f64; 256]; + for k in 0..256 { + let mut z = b2[k] as f64; + for j in 0..dim { + z += h[j] * w2[j * 256 + k] as f64; + } + logits[k] = z; + } + let max = logits.iter().cloned().fold(f64::NEG_INFINITY, f64::max); + let sum: f64 = logits.iter().map(|z| (z - max).exp()).sum(); + let p_y = (logits[y] - max).exp() / sum; + let expected = -p_y.log2(); + + let got = model.heldout_bits(&[x as u8, y as u8]); + assert!( + (got - expected).abs() < 1e-6, + "got {got}, expected {expected}" + ); + } + + #[test] + fn test_short_input_has_zero_bits() { + let model = TinyByteModel::new(16, 1); + assert_eq!(model.heldout_bits(&[]), 0.0); + assert_eq!(model.heldout_bits(b"a"), 0.0); + } + + #[test] + fn test_same_seed_gives_identical_params() { + let text = repetitive_text(1024); + let mut a = TinyByteModel::new(16, 5); + let mut b = TinyByteModel::new(16, 5); + assert_eq!(a.params(), b.params()); + a.train_steps(&text, 50, 9); + b.train_steps(&text, 50, 9); + assert_eq!(a.params(), b.params()); + + let mut c = TinyByteModel::new(16, 5); + c.train_steps(&text, 50, 10); + assert_ne!(a.params(), c.params()); + } + + #[test] + fn test_set_params_round_trip() { + let a = TinyByteModel::new(16, 1); + let mut b = TinyByteModel::new(16, 2); + assert_ne!(a.params(), b.params()); + b.set_params(&a.params()); + assert_eq!(a.params(), b.params()); + assert_eq!(a.heldout_bits(b"abc"), b.heldout_bits(b"abc")); + } +} diff --git a/crates/oh-core/src/lib.rs b/crates/oh-core/src/lib.rs new file mode 100644 index 0000000..f10c352 --- /dev/null +++ b/crates/oh-core/src/lib.rs @@ -0,0 +1,16 @@ +//! Open Humanity volunteer compute, version 0. +//! +//! Volunteers train a shared model locally and return only a signed, +//! compressed update. The coordinator scores each update in Turings +//! (held-out bits saved) and credits the volunteer. Design notes live in +//! `docs/design/volunteer-compute-v0.md`. + +pub mod learner; +pub mod scoring; +pub mod update; +pub mod wire; + +pub use learner::*; +pub use scoring::*; +pub use update::*; +pub use wire::*; diff --git a/crates/oh-core/src/scoring.rs b/crates/oh-core/src/scoring.rs new file mode 100644 index 0000000..e371893 --- /dev/null +++ b/crates/oh-core/src/scoring.rs @@ -0,0 +1,428 @@ +//! Scoring updates in Turings, and the append-only ledger of credits. +//! +//! T = held-out bits with the base model minus held-out bits with the base +//! plus the delta. The held-out bytes live only inside `Scorer` on the +//! coordinator; they are never put on the wire. + +use crate::learner::Learner; +use crate::update::apply; +use serde::{Deserialize, Serialize}; +use std::collections::{HashMap, HashSet}; +use std::fs::{File, OpenOptions}; +use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write}; +use std::path::Path; + +/// Scores deltas against a sealed held-out set. +pub struct Scorer { + pub base: Vec, + pub heldout: Vec, + /// Smallest score, in bits, that counts as an improvement. + pub epsilon: f64, +} + +impl Scorer { + /// Turings earned by `delta`: bits(base) - bits(base + delta) on the + /// held-out set. The model's own parameters are restored afterwards. + /// + /// Panics if `delta` or `base` does not match the model's parameter + /// count; validate untrusted deltas first. + pub fn score(&self, model: &mut L, delta: &[f32]) -> f64 { + let saved = model.params(); + model.set_params(&self.base); + let before = model.heldout_bits(&self.heldout); + model.set_params(&apply(&self.base, delta)); + let after = model.heldout_bits(&self.heldout); + model.set_params(&saved); + before - after + } + + /// True when `t` is a finite score above `epsilon`. + pub fn accept(&self, t: f64) -> bool { + t.is_finite() && t > self.epsilon + } +} + +#[derive(Debug, thiserror::Error)] +pub enum LedgerError { + #[error("ledger i/o error: {0}")] + Io(#[from] std::io::Error), + #[error("ledger line {line} is unreadable: {reason}")] + Corrupt { line: usize, reason: String }, + #[error("only finite, positive Turings can be credited (got {0})")] + InvalidTurings(f64), +} + +/// One line of the ledger file. Byte fields are lowercase hex. +#[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +enum Entry { + /// A delta digest the coordinator has already scored in this run. + Seen { + run_id: String, + delta_digest: String, + }, + /// Turings credited to a volunteer for one accepted update. + Credit { + node_pubkey: String, + turings: f64, + receipt_digest: String, + }, + /// An update that was scored or checked and turned down. It earns + /// nothing; the line is kept so every receipt sent has a ledger record. + Rejected { + node_pubkey: String, + /// `None` when the score was not a finite number. + turings: Option, + reason: String, + receipt_digest: String, + }, +} + +/// Append-only JSON Lines ledger. Opening an existing file replays it, so +/// totals and replay protection survive a restart. +pub struct Ledger { + file: File, + totals: HashMap<[u8; 32], f64>, + /// BLAKE3 of run id and delta digest, so protection is per run. + seen: HashSet<[u8; 32]>, +} + +impl Ledger { + /// Opens or creates the ledger at `path`. A file whose last line was + /// cut short (no final newline) is refused rather than appended to. + pub fn open(path: &Path) -> Result { + let mut file = OpenOptions::new() + .read(true) + .append(true) + .create(true) + .open(path)?; + let mut ledger = Self { + file: file.try_clone()?, + totals: HashMap::new(), + seen: HashSet::new(), + }; + + let len = file.seek(SeekFrom::End(0))?; + if len > 0 { + file.seek(SeekFrom::Start(len - 1))?; + let mut last = [0u8; 1]; + file.read_exact(&mut last)?; + if last[0] != b'\n' { + return Err(LedgerError::Corrupt { + line: 0, + reason: "file does not end with a newline".into(), + }); + } + } + file.seek(SeekFrom::Start(0))?; + for (i, line) in BufReader::new(&file).lines().enumerate() { + let line = line?; + let corrupt = |reason: String| LedgerError::Corrupt { + line: i + 1, + reason, + }; + let entry: Entry = serde_json::from_str(&line).map_err(|e| corrupt(e.to_string()))?; + match entry { + Entry::Seen { + run_id, + delta_digest, + } => { + let run = from_hex::<16>(&run_id).ok_or_else(|| corrupt("run_id".into()))?; + let delta = from_hex::<32>(&delta_digest) + .ok_or_else(|| corrupt("delta_digest".into()))?; + ledger.seen.insert(seen_key(&run, &delta)); + } + Entry::Credit { + node_pubkey, + turings, + .. + } => { + let node = from_hex::<32>(&node_pubkey) + .ok_or_else(|| corrupt("node_pubkey".into()))?; + *ledger.totals.entry(node).or_insert(0.0) += turings; + } + Entry::Rejected { node_pubkey, .. } => { + from_hex::<32>(&node_pubkey).ok_or_else(|| corrupt("node_pubkey".into()))?; + } + } + } + Ok(ledger) + } + + /// Records that `delta_digest` was submitted in `run_id`. Returns + /// `false`, and writes nothing, if it was already recorded: a replay. + pub fn mark_seen( + &mut self, + run_id: &[u8; 16], + delta_digest: &[u8; 32], + ) -> Result { + let key = seen_key(run_id, delta_digest); + if self.seen.contains(&key) { + return Ok(false); + } + self.append(&Entry::Seen { + run_id: to_hex(run_id), + delta_digest: to_hex(delta_digest), + })?; + self.seen.insert(key); + Ok(true) + } + + pub fn is_seen(&self, run_id: &[u8; 16], delta_digest: &[u8; 32]) -> bool { + self.seen.contains(&seen_key(run_id, delta_digest)) + } + + /// Credits `turings` to `node_pubkey` for the receipt `receipt_digest`. + pub fn credit( + &mut self, + node_pubkey: &[u8; 32], + turings: f64, + receipt_digest: &[u8; 32], + ) -> Result<(), LedgerError> { + if !(turings.is_finite() && turings > 0.0) { + return Err(LedgerError::InvalidTurings(turings)); + } + self.append(&Entry::Credit { + node_pubkey: to_hex(node_pubkey), + turings, + receipt_digest: to_hex(receipt_digest), + })?; + *self.totals.entry(*node_pubkey).or_insert(0.0) += turings; + Ok(()) + } + + /// Records a rejected update for the receipt `receipt_digest`. Nothing + /// is credited; totals do not change. + pub fn record_rejection( + &mut self, + node_pubkey: &[u8; 32], + turings: f64, + reason: &str, + receipt_digest: &[u8; 32], + ) -> Result<(), LedgerError> { + self.append(&Entry::Rejected { + node_pubkey: to_hex(node_pubkey), + turings: turings.is_finite().then_some(turings), + reason: reason.to_string(), + receipt_digest: to_hex(receipt_digest), + }) + } + + /// Total Turings credited to `node_pubkey`; zero if never credited. + pub fn total(&self, node_pubkey: &[u8; 32]) -> f64 { + self.totals.get(node_pubkey).copied().unwrap_or(0.0) + } + + fn append(&mut self, entry: &Entry) -> Result<(), LedgerError> { + let mut line = serde_json::to_string(entry).expect("ledger entries always encode"); + line.push('\n'); + self.file.write_all(line.as_bytes())?; + self.file.sync_data()?; + Ok(()) + } +} + +fn seen_key(run_id: &[u8; 16], delta_digest: &[u8; 32]) -> [u8; 32] { + let mut hasher = blake3::Hasher::new(); + hasher.update(run_id); + hasher.update(delta_digest); + *hasher.finalize().as_bytes() +} + +fn to_hex(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +fn from_hex(s: &str) -> Option<[u8; N]> { + if s.len() != N * 2 || !s.is_ascii() { + return None; + } + let mut out = [0u8; N]; + for (i, byte) in out.iter_mut().enumerate() { + *byte = u8::from_str_radix(&s[i * 2..i * 2 + 2], 16).ok()?; + } + Some(out) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::learner::{SplitMix64, TinyByteModel}; + use crate::update::{pseudo_gradient, CompressedDelta}; + use std::path::PathBuf; + + const TRAIN: &str = + "the cat sat on the mat. the dog sat on the log. a cat and a dog met on the mat. "; + const HELDOUT: &str = + "the dog sat on the mat. the cat met a dog on the log. a dog and a cat sat. "; + + fn corpus(text: &str, n: usize) -> Vec { + text.bytes().cycle().take(n).collect() + } + + fn scorer(base: Vec) -> Scorer { + Scorer { + base, + heldout: HELDOUT.as_bytes().to_vec(), + epsilon: 1e-6, + } + } + + fn temp_path(name: &str) -> PathBuf { + let path = + std::env::temp_dir().join(format!("oh-core-{}-{}.jsonl", std::process::id(), name)); + let _ = std::fs::remove_file(&path); + path + } + + /// A volunteer trains on text like the held-out set and sends a 5% delta. + #[test] + fn test_honest_delta_scores_positive() { + let mut model = TinyByteModel::new(16, 1); + let base = model.params(); + model.train_steps(&corpus(TRAIN, 2048), 400, 2); + let delta = pseudo_gradient(&base, &model.params()); + let sent = CompressedDelta::compress(&delta, 0.05).decompress(); + + let s = scorer(base); + let t_full = s.score(&mut model, &delta); + let t_sent = s.score(&mut model, &sent); + eprintln!( + "honest delta: {t_full:.2} Turings dense, {t_sent:.2} Turings after 5% compression" + ); + assert!(s.accept(t_full), "dense delta scored {t_full}"); + assert!(s.accept(t_sent), "compressed delta scored {t_sent}"); + } + + #[test] + fn test_zero_delta_scores_zero_and_is_rejected() { + let mut model = TinyByteModel::new(16, 1); + let s = scorer(model.params()); + let t = s.score(&mut model, &vec![0.0; s.base.len()]); + assert_eq!(t, 0.0); + assert!(!s.accept(t)); + } + + #[test] + fn test_random_delta_scores_negative() { + let mut model = TinyByteModel::new(16, 1); + model.train_steps(&corpus(TRAIN, 2048), 400, 2); + let s = scorer(model.params()); + let mut rng = SplitMix64::new(99); + let noise: Vec = (0..s.base.len()) + .map(|_| rng.next_signed_unit() * 0.5) + .collect(); + let t = s.score(&mut model, &noise); + eprintln!("random delta: {t:.2} Turings"); + assert!(t < 0.0, "random delta scored {t}"); + assert!(!s.accept(t)); + } + + #[test] + fn test_score_restores_the_model() { + let mut model = TinyByteModel::new(16, 4); + let own = model.params(); + let s = scorer(TinyByteModel::new(16, 5).params()); + let _ = s.score(&mut model, &vec![0.01; own.len()]); + assert_eq!(model.params(), own); + } + + #[test] + fn test_nan_score_is_rejected() { + let s = scorer(Vec::new()); + assert!(!s.accept(f64::NAN)); + assert!(!s.accept(f64::INFINITY)); + assert!(s.accept(0.5)); + } + + #[test] + fn test_replayed_delta_digest_is_rejected() { + let path = temp_path("replay"); + let (run_a, run_b, digest) = ([1u8; 16], [2u8; 16], [7u8; 32]); + { + let mut ledger = Ledger::open(&path).unwrap(); + assert!(ledger.mark_seen(&run_a, &digest).unwrap()); + assert!( + !ledger.mark_seen(&run_a, &digest).unwrap(), + "replay must be refused" + ); + assert!( + ledger.mark_seen(&run_b, &digest).unwrap(), + "other runs are separate" + ); + } + let mut reopened = Ledger::open(&path).unwrap(); + assert!(reopened.is_seen(&run_a, &digest)); + assert!( + !reopened.mark_seen(&run_a, &digest).unwrap(), + "replay refused after reopen" + ); + std::fs::remove_file(&path).unwrap(); + } + + #[test] + fn test_ledger_totals_sum_and_survive_reopen() { + let path = temp_path("totals"); + let (alice, bob) = ([0xAA; 32], [0xBB; 32]); + { + let mut ledger = Ledger::open(&path).unwrap(); + ledger.credit(&alice, 1.5, &[1; 32]).unwrap(); + ledger.credit(&bob, 0.25, &[2; 32]).unwrap(); + ledger.credit(&alice, 2.75, &[3; 32]).unwrap(); + assert_eq!(ledger.total(&alice), 4.25); + assert_eq!(ledger.total(&bob), 0.25); + assert_eq!(ledger.total(&[0; 32]), 0.0); + assert!(matches!( + ledger.credit(&bob, 0.0, &[4; 32]), + Err(LedgerError::InvalidTurings(_)) + )); + assert!(matches!( + ledger.credit(&bob, f64::NAN, &[4; 32]), + Err(LedgerError::InvalidTurings(_)) + )); + } + let mut reopened = Ledger::open(&path).unwrap(); + assert_eq!(reopened.total(&alice), 4.25); + assert_eq!(reopened.total(&bob), 0.25); + reopened.credit(&bob, 1.0, &[5; 32]).unwrap(); + assert_eq!(reopened.total(&bob), 1.25); + + let lines = std::fs::read_to_string(&path).unwrap(); + assert_eq!(lines.lines().count(), 4, "append-only: one line per credit"); + std::fs::remove_file(&path).unwrap(); + } + + #[test] + fn test_rejections_are_recorded_but_never_credited() { + let path = temp_path("rejected"); + let node = [0xCC; 32]; + { + let mut ledger = Ledger::open(&path).unwrap(); + ledger.credit(&node, 2.0, &[1; 32]).unwrap(); + ledger + .record_rejection(&node, -3.5, "did not lower held-out bits", &[2; 32]) + .unwrap(); + ledger + .record_rejection(&node, f64::NAN, "score was not a number", &[3; 32]) + .unwrap(); + assert_eq!(ledger.total(&node), 2.0); + } + let text = std::fs::read_to_string(&path).unwrap(); + assert_eq!(text.matches("\"kind\":\"rejected\"").count(), 2); + assert!(text.contains("\"turings\":null")); + let reopened = Ledger::open(&path).unwrap(); + assert_eq!(reopened.total(&node), 2.0); + std::fs::remove_file(&path).unwrap(); + } + + #[test] + fn test_torn_last_line_is_refused() { + let path = temp_path("torn"); + std::fs::write(&path, "{\"kind\":\"credit\",\"node_pub").unwrap(); + assert!(matches!( + Ledger::open(&path), + Err(LedgerError::Corrupt { .. }) + )); + std::fs::remove_file(&path).unwrap(); + } +} diff --git a/crates/oh-core/src/update.rs b/crates/oh-core/src/update.rs new file mode 100644 index 0000000..17612c9 --- /dev/null +++ b/crates/oh-core/src/update.rs @@ -0,0 +1,239 @@ +//! Updates: the change a volunteer made to the model, and its compressed form. +//! +//! A volunteer sends back `after - before` (the pseudo-gradient), keeping only +//! the largest entries by size and storing each as an 8-bit integer times one +//! shared scale. This is the top-k plus quantization idea from SparseLoCo, +//! without its error feedback. + +use serde::{Deserialize, Serialize}; + +/// The change training made: `after - before`, entry by entry. +/// +/// Panics if the two vectors differ in length. +pub fn pseudo_gradient(before: &[f32], after: &[f32]) -> Vec { + assert_eq!( + before.len(), + after.len(), + "parameter vector length mismatch" + ); + after.iter().zip(before).map(|(a, b)| a - b).collect() +} + +/// `base + delta`, entry by entry. +/// +/// Panics if the two vectors differ in length; check untrusted deltas with +/// `CompressedDelta::validate` first. +pub fn apply(base: &[f32], delta: &[f32]) -> Vec { + assert_eq!(base.len(), delta.len(), "parameter vector length mismatch"); + base.iter().zip(delta).map(|(b, d)| b + d).collect() +} + +/// A sparse, 8-bit copy of a dense delta. +/// +/// Entry `indices[i]` of the dense delta is `values[i] as f32 * scale`; every +/// other entry is zero. `indices` is sorted and has no repeats. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct CompressedDelta { + pub len: u32, + pub indices: Vec, + pub values: Vec, + pub scale: f32, +} + +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum DeltaError { + #[error("malformed delta: {0}")] + Malformed(&'static str), +} + +impl CompressedDelta { + /// Keeps the `keep_fraction` largest entries by magnitude (at least one + /// when the fraction is above zero) and quantizes them linearly with + /// `scale = max |kept| / 127`. Zero, NaN and infinite entries are never kept. + pub fn compress(delta: &[f32], keep_fraction: f64) -> Self { + let frac = keep_fraction.clamp(0.0, 1.0); + let mut candidates: Vec = (0..delta.len() as u32) + .filter(|&i| { + let v = delta[i as usize]; + v.is_finite() && v != 0.0 + }) + .collect(); + + let mut k = (delta.len() as f64 * frac).round() as usize; + if frac > 0.0 { + k = k.max(1); + } + let k = k.min(candidates.len()); + + if k < candidates.len() { + // Largest magnitude first; ties broken by index so the result is stable. + let by_size = |a: &u32, b: &u32| { + let (va, vb) = (delta[*a as usize].abs(), delta[*b as usize].abs()); + vb.total_cmp(&va).then(a.cmp(b)) + }; + candidates.select_nth_unstable_by(k, by_size); + candidates.truncate(k); + } + candidates.sort_unstable(); + + let max = candidates + .iter() + .map(|&i| delta[i as usize].abs()) + .fold(0.0f32, f32::max); + let scale = if max > 0.0 { max / 127.0 } else { 0.0 }; + let values = candidates + .iter() + .map(|&i| (delta[i as usize] / scale).round().clamp(-127.0, 127.0) as i8) + .collect(); + + Self { + len: delta.len() as u32, + indices: candidates, + values, + scale, + } + } + + /// Rebuilds the dense delta. Entries whose index is out of range, and + /// indices without a matching value, are skipped; call `validate` first + /// on anything that came over the network. + pub fn decompress(&self) -> Vec { + let mut out = vec![0f32; self.len as usize]; + for (&i, &v) in self.indices.iter().zip(&self.values) { + if let Some(slot) = out.get_mut(i as usize) { + *slot = v as f32 * self.scale; + } + } + out + } + + /// Checks a delta received from someone else before it is used. + pub fn validate(&self, expected_len: usize) -> Result<(), DeltaError> { + if self.len as usize != expected_len { + return Err(DeltaError::Malformed("length does not match the model")); + } + if self.indices.len() != self.values.len() { + return Err(DeltaError::Malformed("indices and values differ in count")); + } + if !self.scale.is_finite() || self.scale < 0.0 { + return Err(DeltaError::Malformed( + "scale is not a finite non-negative number", + )); + } + if !self.indices.windows(2).all(|w| w[0] < w[1]) { + return Err(DeltaError::Malformed("indices are not strictly increasing")); + } + if self.indices.last().is_some_and(|&i| i >= self.len) { + return Err(DeltaError::Malformed("index out of range")); + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::learner::SplitMix64; + + fn random_delta(n: usize, seed: u64) -> Vec { + let mut rng = SplitMix64::new(seed); + (0..n).map(|_| rng.next_signed_unit() * 0.01).collect() + } + + #[test] + fn test_pseudo_gradient_is_after_minus_before() { + let before = [1.0f32, 2.0, 3.0]; + let after = [1.5f32, 2.0, 2.0]; + assert_eq!(pseudo_gradient(&before, &after), vec![0.5, 0.0, -1.0]); + assert_eq!( + apply(&before, &pseudo_gradient(&before, &after)), + after.to_vec() + ); + } + + #[test] + fn test_kept_values_within_half_a_step() { + let delta = random_delta(10_000, 1); + let c = CompressedDelta::compress(&delta, 0.05); + assert_eq!(c.len, 10_000); + assert_eq!(c.indices.len(), 500); + assert_eq!(c.values.len(), 500); + assert!( + c.indices.windows(2).all(|w| w[0] < w[1]), + "indices sorted and unique" + ); + + let out = c.decompress(); + assert_eq!(out.len(), 10_000); + let kept: std::collections::HashSet = c.indices.iter().copied().collect(); + for (i, (orig, got)) in delta.iter().zip(&out).enumerate() { + if kept.contains(&(i as u32)) { + assert!( + (orig - got).abs() <= c.scale / 2.0 + 1e-9, + "index {i}: {orig} vs {got}, scale {}", + c.scale + ); + } else { + assert_eq!(*got, 0.0); + } + } + } + + #[test] + fn test_keeps_the_largest_entries() { + let delta = [0.1f32, -5.0, 0.2, 3.0, -0.3]; + let c = CompressedDelta::compress(&delta, 0.4); + assert_eq!(c.indices, vec![1, 3]); + assert_eq!(c.values, vec![-127, 76]); + } + + #[test] + fn test_compression_ratio_at_five_percent() { + let delta = random_delta(10_000, 2); + let c = CompressedDelta::compress(&delta, 0.05); + let dense = bincode::serialize(&delta).unwrap().len(); + let sparse = bincode::serialize(&c).unwrap().len(); + let ratio = dense as f64 / sparse as f64; + eprintln!("compression: {dense} bytes dense, {sparse} bytes sent, {ratio:.1}x"); + assert!(ratio >= 10.0, "ratio {ratio:.2} ({dense} / {sparse} bytes)"); + } + + #[test] + fn test_zero_delta_compresses_to_nothing() { + let c = CompressedDelta::compress(&vec![0.0f32; 1000], 0.05); + assert!(c.indices.is_empty()); + assert!(c.values.is_empty()); + assert_eq!(c.decompress(), vec![0.0f32; 1000]); + } + + #[test] + fn test_non_finite_entries_are_dropped() { + let delta = [f32::NAN, 1.0, f32::INFINITY, -2.0]; + let c = CompressedDelta::compress(&delta, 1.0); + assert_eq!(c.indices, vec![1, 3]); + assert!(c.decompress().iter().all(|v| v.is_finite())); + } + + #[test] + fn test_validate_rejects_malformed_deltas() { + let good = CompressedDelta::compress(&random_delta(100, 3), 0.1); + assert!(good.validate(100).is_ok()); + assert!(good.validate(99).is_err()); + + let mut out_of_range = good.clone(); + *out_of_range.indices.last_mut().unwrap() = 100; + assert!(out_of_range.validate(100).is_err()); + + let mut unsorted = good.clone(); + unsorted.indices.swap(0, 1); + assert!(unsorted.validate(100).is_err()); + + let mut short_values = good.clone(); + short_values.values.pop(); + assert!(short_values.validate(100).is_err()); + + let mut bad_scale = good.clone(); + bad_scale.scale = f32::NAN; + assert!(bad_scale.validate(100).is_err()); + } +} diff --git a/crates/oh-core/src/wire.rs b/crates/oh-core/src/wire.rs new file mode 100644 index 0000000..f27d7d4 --- /dev/null +++ b/crates/oh-core/src/wire.rs @@ -0,0 +1,474 @@ +//! Wire formats for volunteer compute: work units, updates, receipts, and +//! the length-prefixed frames that carry them over TCP. +//! +//! Signing bytes for each message are the bincode bytes of the message with +//! its `sig` field emptied. Signatures are Ed25519 from beacon-core; digests +//! are BLAKE3. + +use crate::update::CompressedDelta; +use beacon_core::{verify_signature, Keypair}; +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; +use std::io::{Read, Write}; + +/// Largest frame payload either side will send or accept. +pub const MAX_FRAME_BYTES: u32 = 64 * 1024 * 1024; + +#[derive(Debug, thiserror::Error)] +pub enum WireError { + #[error("i/o error: {0}")] + Io(#[from] std::io::Error), + #[error("frame of {0} bytes exceeds the {MAX_FRAME_BYTES} byte cap")] + FrameTooLarge(u64), + #[error("could not decode message: {0}")] + Decode(String), + #[error("signature missing or invalid")] + BadSignature, + #[error("signed by an unexpected key")] + WrongSigner, + #[error("delta digest does not match the delta")] + DeltaDigestMismatch, + #[error("update does not belong to this work unit")] + WorkUnitMismatch, +} + +/// The coordinator's instruction: start from the model whose digest is +/// `base_digest` and train for `steps` steps on `shard`. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct WorkUnit { + pub run_id: [u8; 16], + pub round: u32, + pub base_digest: [u8; 32], + pub shard: Vec, + pub steps: u32, + pub coordinator_pubkey: [u8; 32], + pub sig: Vec, +} + +/// A volunteer's answer to one work unit: the compressed change, never the data. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct Update { + pub run_id: [u8; 16], + pub round: u32, + pub base_digest: [u8; 32], + /// `WorkUnit::digest` of the work unit this answers. + pub work_unit_id: [u8; 32], + pub node_pubkey: [u8; 32], + pub delta: CompressedDelta, + /// `delta_digest(&delta)`. + pub delta_digest: [u8; 32], + pub sig: Vec, +} + +/// The coordinator's signed verdict on one update. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct Receipt { + pub run_id: [u8; 16], + pub round: u32, + /// `Update::digest` of the scored update. + pub update_digest: [u8; 32], + pub turings: f64, + pub accepted: bool, + pub coordinator_pubkey: [u8; 32], + pub sig: Vec, +} + +/// The starting model, sent just before a work unit. It is not signed on its +/// own: the signed work unit pins it through `base_digest`. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct BaseModel { + /// `TinyByteModel` width, so the volunteer can build the right shape. + pub dim: u32, + pub params: Vec, +} + +impl BaseModel { + /// `params_digest` of the parameters; compare with `WorkUnit::base_digest`. + pub fn digest(&self) -> [u8; 32] { + params_digest(&self.params) + } +} + +/// BLAKE3 of the parameters as little-endian f32 bytes. +pub fn params_digest(params: &[f32]) -> [u8; 32] { + let mut hasher = blake3::Hasher::new(); + for p in params { + hasher.update(&p.to_le_bytes()); + } + *hasher.finalize().as_bytes() +} + +/// BLAKE3 of the bincode bytes of a compressed delta. +pub fn delta_digest(delta: &CompressedDelta) -> [u8; 32] { + *blake3::hash(&encode(delta)).as_bytes() +} + +fn encode(value: &T) -> Vec { + // Plain structs of integers, floats and vectors always encode. + bincode::serialize(value).expect("bincode encodes plain structs") +} + +fn check_sig(pubkey: &[u8; 32], message: &[u8], sig: &[u8]) -> Result<(), WireError> { + let sig: [u8; 64] = sig.try_into().map_err(|_| WireError::BadSignature)?; + verify_signature(pubkey, message, &sig).map_err(|_| WireError::BadSignature) +} + +/// Shared signing logic: every message is signed over itself with `sig` emptied. +macro_rules! signed_message { + ($ty:ty, $signer:ident) => { + impl $ty { + /// Bincode bytes of this message with `sig` emptied. + pub fn signing_bytes(&self) -> Vec { + let mut unsigned = self.clone(); + unsigned.sig = Vec::new(); + encode(&unsigned) + } + + /// BLAKE3 of `signing_bytes`. + pub fn digest(&self) -> [u8; 32] { + *blake3::hash(&self.signing_bytes()).as_bytes() + } + + /// Sets the signer's public key, then signs. + pub fn sign(&mut self, keypair: &Keypair) { + self.$signer = keypair.pubkey_bytes(); + self.sig = keypair.sign(&self.signing_bytes()).to_vec(); + } + + /// Checks the signature against the key named in the message. + /// This proves integrity only; use `verify_from` to also pin who signed. + pub fn verify(&self) -> Result<(), WireError> { + check_sig(&self.$signer, &self.signing_bytes(), &self.sig)?; + self.check_contents() + } + + /// `verify`, and the signer must be `expected`. + pub fn verify_from(&self, expected: &[u8; 32]) -> Result<(), WireError> { + if &self.$signer != expected { + return Err(WireError::WrongSigner); + } + self.verify() + } + } + }; +} + +signed_message!(WorkUnit, coordinator_pubkey); +signed_message!(Update, node_pubkey); +signed_message!(Receipt, coordinator_pubkey); + +impl WorkUnit { + pub fn new_signed( + keypair: &Keypair, + run_id: [u8; 16], + round: u32, + base_digest: [u8; 32], + shard: Vec, + steps: u32, + ) -> Self { + let mut unit = Self { + run_id, + round, + base_digest, + shard, + steps, + coordinator_pubkey: [0; 32], + sig: Vec::new(), + }; + unit.sign(keypair); + unit + } + + fn check_contents(&self) -> Result<(), WireError> { + Ok(()) + } +} + +impl Update { + /// Builds and signs the answer to `unit`, copying its run, round, base + /// and id so the update cannot be moved to another work unit. + pub fn new_signed(keypair: &Keypair, unit: &WorkUnit, delta: CompressedDelta) -> Self { + let mut update = Self { + run_id: unit.run_id, + round: unit.round, + base_digest: unit.base_digest, + work_unit_id: unit.digest(), + node_pubkey: [0; 32], + delta_digest: delta_digest(&delta), + delta, + sig: Vec::new(), + }; + update.sign(keypair); + update + } + + /// `verify`, and the update must answer exactly `unit`. + pub fn verify_for(&self, unit: &WorkUnit) -> Result<(), WireError> { + if self.run_id != unit.run_id + || self.round != unit.round + || self.base_digest != unit.base_digest + || self.work_unit_id != unit.digest() + { + return Err(WireError::WorkUnitMismatch); + } + self.verify() + } + + fn check_contents(&self) -> Result<(), WireError> { + if delta_digest(&self.delta) != self.delta_digest { + return Err(WireError::DeltaDigestMismatch); + } + Ok(()) + } +} + +impl Receipt { + pub fn new_signed(keypair: &Keypair, update: &Update, turings: f64, accepted: bool) -> Self { + let mut receipt = Self { + run_id: update.run_id, + round: update.round, + update_digest: update.digest(), + turings, + accepted, + coordinator_pubkey: [0; 32], + sig: Vec::new(), + }; + receipt.sign(keypair); + receipt + } + + fn check_contents(&self) -> Result<(), WireError> { + Ok(()) + } +} + +/// Writes one frame: a 4-byte big-endian length, then the payload. +pub fn write_frame(w: &mut W, payload: &[u8]) -> Result<(), WireError> { + if payload.len() as u64 > MAX_FRAME_BYTES as u64 { + return Err(WireError::FrameTooLarge(payload.len() as u64)); + } + w.write_all(&(payload.len() as u32).to_be_bytes())?; + w.write_all(payload)?; + w.flush()?; + Ok(()) +} + +/// Reads one frame. The length is checked against the cap before any +/// buffer is set aside for the payload. +pub fn read_frame(r: &mut R) -> Result, WireError> { + let mut header = [0u8; 4]; + r.read_exact(&mut header)?; + let len = u32::from_be_bytes(header); + if len > MAX_FRAME_BYTES { + return Err(WireError::FrameTooLarge(len as u64)); + } + let mut payload = vec![0u8; len as usize]; + r.read_exact(&mut payload)?; + Ok(payload) +} + +/// Encodes `msg` with bincode and writes it as one frame. +pub fn send(w: &mut W, msg: &T) -> Result<(), WireError> { + write_frame(w, &encode(msg)) +} + +/// Reads one frame and decodes it with bincode. +pub fn recv(r: &mut R) -> Result { + let payload = read_frame(r)?; + bincode::deserialize(&payload).map_err(|e| WireError::Decode(e.to_string())) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Cursor; + + fn unit(kp: &Keypair) -> WorkUnit { + WorkUnit::new_signed(kp, [7; 16], 3, [9; 32], b"shard bytes".to_vec(), 100) + } + + fn delta() -> CompressedDelta { + CompressedDelta::compress(&[0.0, 0.5, -1.0, 0.25], 0.5) + } + + #[test] + fn test_work_unit_sign_then_verify() { + let kp = Keypair::generate(); + let wu = unit(&kp); + assert_eq!(wu.coordinator_pubkey, kp.pubkey_bytes()); + assert_eq!(wu.sig.len(), 64); + wu.verify().unwrap(); + wu.verify_from(&kp.pubkey_bytes()).unwrap(); + assert!(matches!( + wu.verify_from(&Keypair::generate().pubkey_bytes()), + Err(WireError::WrongSigner) + )); + } + + #[test] + fn test_flipped_shard_byte_fails_verify() { + let kp = Keypair::generate(); + let mut wu = unit(&kp); + wu.shard[0] ^= 1; + assert!(matches!(wu.verify(), Err(WireError::BadSignature))); + } + + #[test] + fn test_missing_or_short_signature_fails_verify() { + let kp = Keypair::generate(); + let mut wu = unit(&kp); + wu.sig.truncate(63); + assert!(matches!(wu.verify(), Err(WireError::BadSignature))); + wu.sig.clear(); + assert!(matches!(wu.verify(), Err(WireError::BadSignature))); + } + + #[test] + fn test_update_binds_to_its_work_unit() { + let coordinator = Keypair::generate(); + let node = Keypair::generate(); + let wu = unit(&coordinator); + let up = Update::new_signed(&node, &wu, delta()); + assert_eq!(up.work_unit_id, wu.digest()); + assert_eq!(up.node_pubkey, node.pubkey_bytes()); + up.verify_for(&wu).unwrap(); + + let other = WorkUnit::new_signed( + &coordinator, + [7; 16], + 4, + [9; 32], + b"shard bytes".to_vec(), + 100, + ); + assert!(matches!( + up.verify_for(&other), + Err(WireError::WorkUnitMismatch) + )); + } + + #[test] + fn test_tampered_update_fails_verify() { + let wu = unit(&Keypair::generate()); + let node = Keypair::generate(); + let up = Update::new_signed(&node, &wu, delta()); + + let mut changed_value = up.clone(); + changed_value.delta.values[0] = changed_value.delta.values[0].wrapping_add(1); + assert!(changed_value.verify().is_err()); + + let mut moved_round = up.clone(); + moved_round.round += 1; + assert!(matches!(moved_round.verify(), Err(WireError::BadSignature))); + + // Re-signing a forged digest still fails: the digest must match the delta. + let mut forged = up.clone(); + forged.delta.values[0] = 0; + forged.sign(&node); + assert!(matches!( + forged.verify(), + Err(WireError::DeltaDigestMismatch) + )); + } + + #[test] + fn test_receipt_flipped_turings_fails_verify() { + let coordinator = Keypair::generate(); + let wu = unit(&coordinator); + let up = Update::new_signed(&Keypair::generate(), &wu, delta()); + let receipt = Receipt::new_signed(&coordinator, &up, 12.5, true); + assert_eq!(receipt.update_digest, up.digest()); + receipt.verify_from(&coordinator.pubkey_bytes()).unwrap(); + + let mut inflated = receipt.clone(); + inflated.turings = f64::from_bits(inflated.turings.to_bits() ^ 1); + assert!(matches!(inflated.verify(), Err(WireError::BadSignature))); + + let mut flipped = receipt.clone(); + flipped.accepted = false; + assert!(matches!(flipped.verify(), Err(WireError::BadSignature))); + } + + #[test] + fn test_digest_ignores_signature() { + let wu = unit(&Keypair::generate()); + let mut resigned = wu.clone(); + resigned.sig = vec![0; 64]; + assert_eq!(wu.digest(), resigned.digest()); + assert_eq!(params_digest(&[1.0, 2.0]), params_digest(&[1.0, 2.0])); + assert_ne!(params_digest(&[1.0, 2.0]), params_digest(&[2.0, 1.0])); + } + + #[test] + fn test_frame_round_trip() { + let kp = Keypair::generate(); + let wu = unit(&kp); + let mut buf = Vec::new(); + send(&mut buf, &wu).unwrap(); + write_frame(&mut buf, b"").unwrap(); + write_frame(&mut buf, b"second").unwrap(); + assert_eq!( + &buf[..4], + &((buf.len() - 4 - 4 - 4 - 6) as u32).to_be_bytes() + ); + + let mut cursor = Cursor::new(buf); + let back: WorkUnit = recv(&mut cursor).unwrap(); + assert_eq!(back, wu); + back.verify().unwrap(); + assert_eq!(read_frame(&mut cursor).unwrap(), b""); + assert_eq!(read_frame(&mut cursor).unwrap(), b"second"); + assert!(matches!(read_frame(&mut cursor), Err(WireError::Io(_)))); + } + + /// Only the 4-byte header is present. Reading the body would hit end of + /// input, so `FrameTooLarge` shows the length was refused before any + /// buffer was set aside or any body read was attempted. + #[test] + fn test_oversize_length_rejected_without_allocating() { + let header = (MAX_FRAME_BYTES + 1).to_be_bytes(); + let mut cursor = Cursor::new(header.to_vec()); + assert!(matches!( + read_frame(&mut cursor), + Err(WireError::FrameTooLarge(n)) if n == MAX_FRAME_BYTES as u64 + 1 + )); + + let mut cursor = Cursor::new(u32::MAX.to_be_bytes().to_vec()); + assert!(matches!( + read_frame(&mut cursor), + Err(WireError::FrameTooLarge(_)) + )); + } + + #[test] + fn test_truncated_frame_is_an_error() { + let mut buf = Vec::new(); + write_frame(&mut buf, b"hello").unwrap(); + buf.truncate(buf.len() - 1); + assert!(matches!( + read_frame(&mut Cursor::new(buf)), + Err(WireError::Io(_)) + )); + } + + #[test] + fn test_garbage_payload_is_a_decode_error() { + let mut buf = Vec::new(); + write_frame(&mut buf, &[1, 2, 3]).unwrap(); + let result: Result = recv(&mut Cursor::new(buf)); + assert!(matches!(result, Err(WireError::Decode(_)))); + } + + #[test] + fn test_base_model_digest_matches_its_params() { + let base = BaseModel { + dim: 2, + params: vec![0.5, -1.0, 2.0], + }; + assert_eq!(base.digest(), params_digest(&[0.5, -1.0, 2.0])); + let mut buf = Vec::new(); + send(&mut buf, &base).unwrap(); + let back: BaseModel = recv(&mut Cursor::new(buf)).unwrap(); + assert_eq!(back, base); + } +} diff --git a/crates/oh-volunteer/Cargo.toml b/crates/oh-volunteer/Cargo.toml new file mode 100644 index 0000000..90d5bf3 --- /dev/null +++ b/crates/oh-volunteer/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "oh-volunteer" +version.workspace = true +edition.workspace = true +license.workspace = true +authors.workspace = true +repository.workspace = true + +[dependencies] +oh-core = { path = "../oh-core" } +beacon-core = { path = "../beacon-core" } +thiserror = "1.0" diff --git a/crates/oh-volunteer/src/lib.rs b/crates/oh-volunteer/src/lib.rs new file mode 100644 index 0000000..493acfb --- /dev/null +++ b/crates/oh-volunteer/src/lib.rs @@ -0,0 +1,175 @@ +//! Volunteer side of Open Humanity volunteer compute, version 0. +//! +//! A volunteer receives the starting model and a signed work unit, trains on +//! the shard on its own machine, and sends back only a signed, compressed +//! update. The shard and the trained parameters stay in this process. + +use beacon_core::Keypair; +use oh_core::{ + pseudo_gradient, recv, send, BaseModel, CompressedDelta, Learner, Receipt, TinyByteModel, + Update, WireError, WorkUnit, +}; +use std::io::{Read, Write}; + +/// Share of parameter changes kept in each update. +pub const DEFAULT_KEEP_FRACTION: f64 = 0.05; + +/// Widest model a volunteer will build. Keeps the parameter count +/// arithmetic far from overflow for any `dim` a coordinator sends. +pub const MAX_DIM: u32 = 4096; + +pub struct Settings { + /// Seeds the order in which training samples the shard. + pub seed: u64, + pub keep_fraction: f64, + /// When set, work units signed by any other key are refused. + pub coordinator_key: Option<[u8; 32]>, +} + +#[derive(Debug, thiserror::Error)] +pub enum VolunteerError { + #[error(transparent)] + Wire(#[from] WireError), + #[error("starting model does not match the work unit: {0}")] + BadBase(&'static str), + #[error("receipt does not answer the update that was sent")] + ReceiptMismatch, +} + +/// What came back for one work unit. +#[derive(Debug, Clone)] +pub struct Outcome { + pub round: u32, + pub accepted: bool, + pub turings: f64, + pub receipt: Receipt, + pub coordinator_key: [u8; 32], +} + +/// Training seed for one round. The coordinator's `--baseline` mode uses +/// the same formula. +pub fn train_seed(seed: u64, round: u32) -> u64 { + seed ^ ((round as u64) << 32) +} + +/// Takes one work unit over `stream`: receive, check, train, send the +/// update, and check the receipt. Nothing is written to `stream` unless the +/// work unit's signature and starting model check out. +pub fn run_once( + stream: &mut S, + keypair: &Keypair, + settings: &Settings, +) -> Result { + let base: BaseModel = recv(stream)?; + let unit: WorkUnit = recv(stream)?; + match &settings.coordinator_key { + Some(key) => unit.verify_from(key)?, + None => unit.verify()?, + } + let update = train(&base, &unit, keypair, settings)?; + send(stream, &update)?; + + let receipt: Receipt = recv(stream)?; + receipt.verify_from(&unit.coordinator_pubkey)?; + if receipt.update_digest != update.digest() + || receipt.run_id != unit.run_id + || receipt.round != unit.round + { + return Err(VolunteerError::ReceiptMismatch); + } + Ok(Outcome { + round: unit.round, + accepted: receipt.accepted, + turings: receipt.turings, + coordinator_key: unit.coordinator_pubkey, + receipt, + }) +} + +/// Trains from `base` on the unit's shard and returns the signed update. +/// Only the compressed parameter change goes into the update. +pub fn train( + base: &BaseModel, + unit: &WorkUnit, + keypair: &Keypair, + settings: &Settings, +) -> Result { + if base.digest() != unit.base_digest { + return Err(VolunteerError::BadBase("digest differs from base_digest")); + } + if base.dim == 0 || base.dim > MAX_DIM { + return Err(VolunteerError::BadBase("dim is zero or too large")); + } + let dim = base.dim as usize; + if TinyByteModel::num_params(dim) != base.params.len() { + return Err(VolunteerError::BadBase("parameter count does not fit dim")); + } + let mut model = TinyByteModel::new(dim, 0); + model.set_params(&base.params); + model.train_steps( + &unit.shard, + unit.steps as usize, + train_seed(settings.seed, unit.round), + ); + let delta = pseudo_gradient(&base.params, &model.params()); + let compressed = CompressedDelta::compress(&delta, settings.keep_fraction); + Ok(Update::new_signed(keypair, unit, compressed)) +} + +/// One plain-English line for the person running the volunteer. +pub fn summary_line(outcome: &Outcome) -> String { + if outcome.accepted { + format!( + "Round {}: update accepted, earned {:.3} Turings", + outcome.round, outcome.turings + ) + } else { + format!("Round {}: update rejected", outcome.round) + } +} + +/// Parses 64 hex characters into a 32-byte key. +pub fn parse_key(hex: &str) -> Option<[u8; 32]> { + if hex.len() != 64 || !hex.is_ascii() { + return None; + } + let mut out = [0u8; 32]; + for (i, byte) in out.iter_mut().enumerate() { + *byte = u8::from_str_radix(&hex[i * 2..i * 2 + 2], 16).ok()?; + } + Some(out) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_key_round_trip() { + let key = Keypair::generate().pubkey_bytes(); + let hex: String = key.iter().map(|b| format!("{b:02x}")).collect(); + assert_eq!(parse_key(&hex), Some(key)); + assert_eq!(parse_key("abc"), None); + assert_eq!(parse_key(&"zz".repeat(32)), None); + } + + #[test] + fn test_summary_line_wording() { + let kp = Keypair::generate(); + let unit = WorkUnit::new_signed(&kp, [0; 16], 4, [0; 32], vec![], 1); + let update = Update::new_signed(&kp, &unit, CompressedDelta::compress(&[1.0], 1.0)); + let mut outcome = Outcome { + round: 4, + accepted: true, + turings: 2.5, + receipt: Receipt::new_signed(&kp, &update, 2.5, true), + coordinator_key: kp.pubkey_bytes(), + }; + assert_eq!( + summary_line(&outcome), + "Round 4: update accepted, earned 2.500 Turings" + ); + outcome.accepted = false; + assert_eq!(summary_line(&outcome), "Round 4: update rejected"); + } +} diff --git a/crates/oh-volunteer/src/main.rs b/crates/oh-volunteer/src/main.rs new file mode 100644 index 0000000..5a06cbe --- /dev/null +++ b/crates/oh-volunteer/src/main.rs @@ -0,0 +1,102 @@ +//! `oh-volunteer`: lend this machine's time to train the shared model. +//! +//! Usage: oh-volunteer --coordinator [--rounds ] [--seed ] +//! [--keep ] [--coordinator-key ] + +use beacon_core::Keypair; +use oh_volunteer::{parse_key, run_once, summary_line, Settings, DEFAULT_KEEP_FRACTION}; +use std::net::TcpStream; +use std::process::ExitCode; +use std::thread; +use std::time::{Duration, Instant}; + +struct Args { + coordinator: String, + rounds: u32, + settings: Settings, +} + +fn parse_args() -> Result { + let mut args = Args { + coordinator: String::new(), + rounds: 1, + settings: Settings { + seed: 1, + keep_fraction: DEFAULT_KEEP_FRACTION, + coordinator_key: None, + }, + }; + let mut it = std::env::args().skip(1); + while let Some(flag) = it.next() { + let value = it.next().ok_or(format!("{flag} needs a value"))?; + let bad = || format!("bad value for {flag}: {value}"); + match flag.as_str() { + "--coordinator" => args.coordinator = value, + "--rounds" => args.rounds = value.parse().map_err(|_| bad())?, + "--seed" => args.settings.seed = value.parse().map_err(|_| bad())?, + "--keep" => { + let f: f64 = value.parse().map_err(|_| bad())?; + if !(f > 0.0 && f <= 1.0) { + return Err(bad()); + } + args.settings.keep_fraction = f; + } + "--coordinator-key" => { + args.settings.coordinator_key = Some(parse_key(&value).ok_or_else(bad)?) + } + _ => return Err(format!("unknown argument {flag}")), + } + } + if args.coordinator.is_empty() { + return Err("--coordinator is required".into()); + } + Ok(args) +} + +/// The coordinator may not be listening yet; keep trying for a minute. +fn connect(addr: &str) -> Result { + let deadline = Instant::now() + Duration::from_secs(60); + loop { + match TcpStream::connect(addr) { + Ok(stream) => { + stream + .set_read_timeout(Some(Duration::from_secs(600))) + .map_err(|e| e.to_string())?; + return Ok(stream); + } + Err(e) if Instant::now() >= deadline => { + return Err(format!("could not reach coordinator at {addr}: {e}")) + } + Err(_) => thread::sleep(Duration::from_millis(200)), + } + } +} + +fn main() -> ExitCode { + let mut args = match parse_args() { + Ok(a) => a, + Err(e) => { + eprintln!("oh-volunteer: {e}"); + return ExitCode::from(2); + } + }; + // The key lives only in memory and is gone when the process exits. + let keypair = Keypair::generate(); + for _ in 0..args.rounds { + let result = connect(&args.coordinator).and_then(|mut stream| { + run_once(&mut stream, &keypair, &args.settings).map_err(|e| e.to_string()) + }); + match result { + Ok(outcome) => { + println!("{}", summary_line(&outcome)); + // Trust the first coordinator key seen for the rest of the run. + args.settings.coordinator_key = Some(outcome.coordinator_key); + } + Err(e) => { + eprintln!("oh-volunteer: work unit failed: {e}"); + return ExitCode::FAILURE; + } + } + } + ExitCode::SUCCESS +} diff --git a/crates/oh-volunteer/tests/data_stays_home.rs b/crates/oh-volunteer/tests/data_stays_home.rs new file mode 100644 index 0000000..1338b3c --- /dev/null +++ b/crates/oh-volunteer/tests/data_stays_home.rs @@ -0,0 +1,222 @@ +//! The volunteer sends back only its signed update, and refuses work it +//! cannot trust. + +use beacon_core::Keypair; +use oh_core::{ + recv, send, BaseModel, Learner, Receipt, TinyByteModel, Update, WireError, WorkUnit, +}; +use oh_volunteer::{run_once, Settings, VolunteerError}; +use std::io::{Cursor, Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::thread; + +/// Random lowercase letters and spaces: every 16-byte run is unique, so a +/// match in outbound bytes can only mean the shard leaked. +fn random_text(n: usize, seed: u64) -> Vec { + let mut state = seed; + (0..n) + .map(|_| { + state = state + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + b"abcdefghijklmnopqrstuvwxyz "[((state >> 33) % 27) as usize] + }) + .collect() +} + +fn contains(haystack: &[u8], needle: &[u8]) -> bool { + haystack.windows(needle.len()).any(|w| w == needle) +} + +fn base_model(dim: usize) -> BaseModel { + BaseModel { + dim: dim as u32, + params: TinyByteModel::new(dim, 1).params(), + } +} + +fn settings() -> Settings { + Settings { + seed: 3, + keep_fraction: 0.05, + coordinator_key: None, + } +} + +/// Passes reads through and keeps a copy of every byte written. +struct Recorder { + inner: S, + sent: Vec, +} + +impl Read for Recorder { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + self.inner.read(buf) + } +} + +impl Write for Recorder { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + let n = self.inner.write(buf)?; + self.sent.extend_from_slice(&buf[..n]); + Ok(n) + } + fn flush(&mut self) -> std::io::Result<()> { + self.inner.flush() + } +} + +/// A scripted coordinator: fixed bytes to read, and a sink for writes. +fn scripted(base: &BaseModel, unit: &WorkUnit) -> Recorder>> { + let mut input = Vec::new(); + send(&mut input, base).unwrap(); + send(&mut input, unit).unwrap(); + Recorder { + inner: Cursor::new(input), + sent: Vec::new(), + } +} + +#[test] +fn test_update_frame_carries_no_shard_bytes() { + let coordinator = Keypair::generate(); + let base = base_model(8); + let shard = random_text(4096, 7); + let unit = WorkUnit::new_signed(&coordinator, [1; 16], 1, base.digest(), shard.clone(), 40); + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let server_unit = unit.clone(); + let server_base = base.clone(); + let server = thread::spawn(move || { + let (mut s, _) = listener.accept().unwrap(); + send(&mut s, &server_base).unwrap(); + send(&mut s, &server_unit).unwrap(); + let update: Update = recv(&mut s).unwrap(); + update.verify_for(&server_unit).unwrap(); + send( + &mut s, + &Receipt::new_signed(&coordinator, &update, 1.5, true), + ) + .unwrap(); + }); + + let mut stream = Recorder { + inner: TcpStream::connect(addr).unwrap(), + sent: Vec::new(), + }; + let outcome = run_once(&mut stream, &Keypair::generate(), &settings()).unwrap(); + server.join().unwrap(); + assert!(outcome.accepted); + assert_eq!(outcome.turings, 1.5); + + // Exactly one frame went out, and it is exactly the encoded update. + let sent = stream.sent; + let len = u32::from_be_bytes(sent[..4].try_into().unwrap()) as usize; + assert_eq!(sent.len(), 4 + len, "one frame and nothing else"); + let update: Update = recv(&mut Cursor::new(&sent)).unwrap(); + let mut reencoded = Vec::new(); + send(&mut reencoded, &update).unwrap(); + assert_eq!( + reencoded, sent, + "the frame holds the update and no extra bytes" + ); + + // The update keeps at most 5% of the parameters, as 8-bit values. + let kept_cap = (TinyByteModel::num_params(8) as f64 * 0.05).round() as usize; + assert!(update.delta.indices.len() <= kept_cap); + + // No 16-byte run of the shard appears in anything sent. + for window in shard.windows(16) { + assert!(!contains(&sent, window), "shard bytes left the volunteer"); + } + // Negative control: the same check does find shard runs in the work unit. + let mut unit_bytes = Vec::new(); + send(&mut unit_bytes, &unit).unwrap(); + assert!(contains(&unit_bytes, &shard[100..116])); +} + +#[test] +fn test_work_unit_with_bad_signature_is_refused() { + let coordinator = Keypair::generate(); + let base = base_model(8); + let mut unit = WorkUnit::new_signed( + &coordinator, + [1; 16], + 1, + base.digest(), + random_text(512, 1), + 10, + ); + unit.shard[0] ^= 1; + + let mut stream = scripted(&base, &unit); + let err = run_once(&mut stream, &Keypair::generate(), &settings()).unwrap_err(); + assert!( + matches!(err, VolunteerError::Wire(WireError::BadSignature)), + "got {err:?}" + ); + assert!(stream.sent.is_empty(), "nothing is sent for a refused unit"); +} + +#[test] +fn test_work_unit_from_unexpected_coordinator_is_refused() { + let base = base_model(8); + let unit = WorkUnit::new_signed( + &Keypair::generate(), + [1; 16], + 1, + base.digest(), + random_text(512, 1), + 10, + ); + let mut pinned = settings(); + pinned.coordinator_key = Some(Keypair::generate().pubkey_bytes()); + + let mut stream = scripted(&base, &unit); + let err = run_once(&mut stream, &Keypair::generate(), &pinned).unwrap_err(); + assert!( + matches!(err, VolunteerError::Wire(WireError::WrongSigner)), + "got {err:?}" + ); + assert!(stream.sent.is_empty()); +} + +#[test] +fn test_base_model_that_does_not_match_the_unit_is_refused() { + let base = base_model(8); + let unit = WorkUnit::new_signed( + &Keypair::generate(), + [1; 16], + 1, + [0; 32], + random_text(512, 1), + 10, + ); + + let mut stream = scripted(&base, &unit); + let err = run_once(&mut stream, &Keypair::generate(), &settings()).unwrap_err(); + assert!(matches!(err, VolunteerError::BadBase(_)), "got {err:?}"); + assert!(stream.sent.is_empty()); +} + +#[test] +fn test_base_model_with_absurd_width_is_refused() { + let base = BaseModel { + dim: u32::MAX, + params: vec![0.0; 4], + }; + let unit = WorkUnit::new_signed( + &Keypair::generate(), + [1; 16], + 1, + base.digest(), + random_text(512, 1), + 10, + ); + + let mut stream = scripted(&base, &unit); + let err = run_once(&mut stream, &Keypair::generate(), &settings()).unwrap_err(); + assert!(matches!(err, VolunteerError::BadBase(_)), "got {err:?}"); + assert!(stream.sent.is_empty()); +} diff --git a/docs/SOVEREIGN_MANIFESTO.md b/docs/SOVEREIGN_MANIFESTO.md index af56b9b..c3a1075 100644 --- a/docs/SOVEREIGN_MANIFESTO.md +++ b/docs/SOVEREIGN_MANIFESTO.md @@ -26,7 +26,7 @@ Every background daemon, process supervisor, memory engine, and API gateway must Software built for human autonomy must respect human privacy as an absolute axiom: - Zero tracking scripts, analytics SDKs, or covert profiling. - The Personal Data Firewall sanitizes file paths and blocks secret keys before any peer interaction. -- The open network never trains on user queries or builds centralized behavioral databases. +- The open network never trains on user queries or builds centralized behavioral databases. The shared model learns only from signed volunteer updates and teaching text, and personal data never leaves the volunteer's machine. ### 4. Zero Plaintext Disk Secrets (Hardware TPM Vault) No `.env`, `.env.*`, or plaintext secret files are permitted on disk in project workspaces. All API tokens, private signing keys, and vault entries must reside in hardware silicon via the Trusted Platform Module (`atlas-vault`) and resolve dynamically in memory with automatic stream redaction. diff --git a/docs/design/volunteer-compute-v0.md b/docs/design/volunteer-compute-v0.md new file mode 100644 index 0000000..3aa99ce --- /dev/null +++ b/docs/design/volunteer-compute-v0.md @@ -0,0 +1,85 @@ +# Volunteer compute, version 0 (design) + +Status: design for the first working slice. Pure Rust, CPU only. + +## Goal + +Let volunteers lend spare computer time to train one open model. Each volunteer trains a copy of the model on their own machine and sends back only a small, signed summary of how the model changed (an "update"). A coordinator checks every update against text the volunteers never see, keeps the updates that make the model better, and credits each volunteer for the improvement they caused. + +Version 0 proves the full loop end to end on one machine (coordinator plus volunteers talking over the local network) with a toy model. The real model plugs in later behind the same interface. + +## Principle change + +Until now the README (line 9) said: "It does not train models. It does not collect user data. Nothing is harvested, and no central model learns from people." `docs/SOVEREIGN_MANIFESTO.md` says the same in its own words: "The open network never trains on user queries or builds centralized behavioral databases." + +Volunteer compute changes the first half of that promise. The new promise is: + +> The model learns only from signed updates and teaching text. Personal data never leaves the volunteer's machine. + +Drake approved this direction in writing on 2026-10-16: "implement the Open Humanity idea: start from the beginning, go all the way through". He gave that approval after these decisions: + +1. Code and model are licensed AGPL-3.0-or-later. +2. No Python and no CUDA: pure Rust, and CPU is fine for version 0. +3. Volunteers train locally and send back only a compressed update. Each update is scored in Turings (defined below). An update scoring zero or less is rejected. The score is credited to the volunteer. +4. Personal data never leaves the volunteer's machine. +5. Build for Drake's one model first (Qwen as teacher later). Open Humanity does not police what others run. +6. Later, a local model generates teaching text that is scored the same way ("AI v1 teaches AI v2"). Not part of version 0. + +The rest of the old promise stands: no user data is collected, and nothing is harvested. + +## Invariants + +1. The held-out text is sealed. It stays on the coordinator and is never placed in any message sent to a volunteer. +2. A volunteer sends back only its update (a sparse, quantized list of parameter changes), a digest of it, its public key and a signature. It never sends training text back. +3. Every work unit is signed by the coordinator. Every update is signed by the volunteer. Every receipt is signed by the coordinator. Anything with a bad signature is dropped. +4. An update names the exact work unit and starting model it came from (by digest), so it cannot be moved to another run or round. +5. The same update (same delta digest) is accepted at most once per run. A replay is rejected. +6. An update is accepted only if its Turing score is above a small positive threshold. +7. The ledger of credits is append-only. + +## Turing definition + +The coordinator keeps a sealed held-out set H of bytes that volunteers never see. For a model with parameters p: + + bits(p) = - sum over each predicted byte b in H of log2 P(b | p) + +That is the number of bits the model needs to encode H; lower is better. For an update with change d applied to the current model p: + + T = bits(p) - bits(p + d) + +T is measured in bits and is called Turings. T > 0 means the update made the model better at predicting text it never trained on. An update is accepted when T > epsilon, a small positive threshold set by the coordinator. + +## Wire formats + +All messages are Rust structs serialized with bincode 1.3 (the same encoding beacon-core uses). Signatures are Ed25519 from beacon-core (`Keypair::sign`, `verify_signature`). Digests are BLAKE3, 32 bytes. + +Signing bytes for a struct are the bincode bytes of that struct with its `sig` field set to empty. The `sig` field then holds the 64-byte signature. + +- `WorkUnit { run_id: [u8;16], round: u32, base_digest: [u8;32], shard: Vec, steps: u32, coordinator_pubkey: [u8;32], sig: Vec }` + The coordinator's instruction: "start from the model whose digest is `base_digest`, train for `steps` steps on `shard`". +- `Update { run_id, round, base_digest, work_unit_id: [u8;32], node_pubkey: [u8;32], delta: CompressedDelta, delta_digest: [u8;32], sig: Vec }` + `work_unit_id` is the BLAKE3 digest of the work unit's signing bytes. `delta_digest` is the BLAKE3 digest of the bincode bytes of `delta`. +- `CompressedDelta { len: u32, indices: Vec, values: Vec, scale: f32 }` + The change to the model, keeping only the largest few percent of entries by size, each stored as an 8-bit integer times `scale`. +- `Receipt { run_id, round, update_digest: [u8;32], turings: f64, accepted: bool, coordinator_pubkey, sig: Vec }` + The coordinator's signed answer: the score and whether the update was kept. + +Over TCP, each message travels in a frame: a 4-byte big-endian length, then that many bytes of bincode. Frames larger than 64 MiB are refused before any memory is set aside for them. + +## Not built in version 0 + +- Differential privacy (mathematical noise that limits what an update can reveal about its training text). +- Secure aggregation (combining updates so the coordinator cannot see any single one). +- GPU training. Version 0 runs on the CPU. +- A real model. The version 0 learner is a toy that predicts the next byte from the previous byte. +- The DiLoCo outer optimizer. Version 0 combines accepted updates with a plain average. +- Sybil resistance. One person can run many keys, and could split one update across them. +- Protection against held-out leakage from repeated scoring. A volunteer who submits many updates and watches the scores could slowly learn about the held-out set. +- Transport encryption. Frames travel in the clear over TCP. Version 0 is meant for loopback only; the shard a volunteer receives and the update it returns are readable by anyone on the path. +- Pinned coordinator identity by default. Unless `--coordinator-key` is given, a volunteer trusts the first coordinator key it sees and holds to it for the rest of its run. +- Protection against slow or silent volunteers. The coordinator waits for one connection per shard and reads each update with a 10 minute timeout, so one stalled volunteer holds up the round. +- A ledger line for dropped updates. Updates with a bad signature, or that answer the wrong work unit, get no receipt and no ledger line; they are only logged. Accepted updates get a credit line and every other update that gets a receipt gets a rejected line. + +## Open questions + +- Why the volunteer runs trail the single-process baseline. At equal total steps the volunteer runs end about 0.33 bits per byte behind (`docs/research/2026-10-16-v0-loopback-results.md`). Likely causes are the plain average of updates trained from the same start and keeping only 5% of each update, but neither has been measured on its own. Next measurement: rerun with `--keep 1.0` (no compression) and with one volunteer, to split the two effects. diff --git a/docs/research/2026-10-16-v0-loopback-results.md b/docs/research/2026-10-16-v0-loopback-results.md new file mode 100644 index 0000000..e4f6a87 --- /dev/null +++ b/docs/research/2026-10-16-v0-loopback-results.md @@ -0,0 +1,64 @@ +# Volunteer compute v0: loopback results + +Date: 2026-10-16. Status: measurement note. Every number below is OBSERVED on the DGX Spark (20 Arm CPU cores, CPU only, release build), from the code at commit `42669cb` on branch `feat/volunteer-compute-v0`. + +## Summary for Drake + +1. The whole loop works on one machine: a coordinator and three volunteer programs talk over the local network, the volunteers train and send back only small signed updates, the coordinator scores each one in Turings, pays out credit, and the shared model gets better every round. +2. Starting from about 8.0 bits per byte (a model that knows nothing), ten rounds brought the shared model to about 4.7 bits per byte on text it never trained on. +3. The same training done in one program, with no network and no shrinking of updates, reached about 4.4 bits per byte. So the volunteer version is about 0.33 to 0.36 bits per byte behind. That gap is the price of averaging and of keeping only 5% of each update; v0 does nothing to close it yet. +4. This proves the plumbing, not model quality. The model is a toy that predicts the next letter from the previous letter, the corpus is one 34 KB file, and everything ran on one computer. + +## What was run + +Command, from the repository root: + + LOG_DIR=/tmp/oh-demo-logs scripts/demo-local.sh + +Defaults used: 10 rounds, 3 volunteers, 500 training steps per work unit, model width 32 (`TinyByteModel`, 17,696 parameters), seeds 1, 2 and 3, 5% of each update kept. Corpus: the repository `LICENSE` file, 34,523 bytes, sha256 starting `0d96a4ff68ad6d4b`. The last 10% (3,453 bytes) is the sealed held-out set; the first 31,070 bytes are split into 3 training shards, one per volunteer, reused every round. + +For each seed the script runs: + +- `oh-coordinator` with 3 `oh-volunteer` processes (volunteer seeds `seed + 1` to `seed + 3`) over loopback TCP. +- `oh-coordinator --baseline`: one model from the same starting seed, trained in one process on the same 3 shards for the same total steps (10 rounds x 3 shards x 500 steps = 15,000 steps), with full updates and no averaging. + +The whole script took about 9 seconds. + +## Numbers + +Final held-out bits per byte (lower is better). The script was run twice. + +| Seed | Start | Volunteers, run 1 | Volunteers, run 2 | Baseline (both runs) | +|---|---|---|---|---| +| 1 | 8.0065 | 4.6533 | 4.6902 | 4.3584 | +| 2 | 8.0000 | 4.7929 | 4.7204 | 4.3878 | +| 3 | 7.9989 | 4.8026 | 4.7585 | 4.4294 | +| Mean | | 4.7496 | 4.7230 | 4.3919 | +| Spread (sample std dev) | | 0.0835 | 0.0342 | 0.0357 | + +Updates accepted (out of 30 per seed): run 1: 16, 14, 15. Run 2: 16, 16, 17. Rounds whose averaged update was applied (out of 10): run 1: 10, 9, 10. Run 2: 10, 10, 10. + +Round by round, seed 1, run 1 (coordinator log): + +| Round | Updates accepted | Turings for the averaged update | Held-out bits per byte after | +|---|---|---|---| +| 1 | 3 | 8211.8 | 5.6276 | +| 2 | 2 | 485.9 | 5.4869 | +| 3 | 2 | 452.5 | 5.3558 | +| 4 | 2 | 399.8 | 5.2400 | +| 5 | 2 | 167.3 | 5.1915 | +| 6 | 1 | 538.0 | 5.0356 | +| 7 | 1 | 389.5 | 4.9228 | +| 8 | 1 | 493.1 | 4.7800 | +| 9 | 1 | 301.4 | 4.6926 | +| 10 | 1 | 135.7 | 4.6533 | + +Baseline, seed 1, held-out bits per byte after each round's 1,500 steps: 5.2518, 4.9753, 4.8497, 4.6910, 4.6329, 4.5255, 4.4080, 4.4396, 4.3617, 4.3584. The baseline has no score gate, so it can get worse in a round (round 8); the volunteer run cannot, because a round is thrown away unless the averaged update lowers held-out bits. + +## Honest reading + +- The pipeline is proven end to end on loopback: signed work units go out, only compressed signed updates come back, every update gets a signed receipt and a Turing score, credits land in the ledger, and held-out bits never rose in any volunteer round. +- The volunteer run is worse than the single-process baseline at equal total steps, by about 0.33 to 0.36 bits per byte on average. Expected causes: the plain mean of 3 updates trained from the same start moves the model about as far as one volunteer's work, and keeping 5% of entries drops most of each update. Which cause dominates is UNVERIFIED; it was not measured separately. +- Run to run, the volunteer numbers change even with the same seeds, because which volunteer gets which shard depends on who connects first. The baseline is deterministic and repeated exactly. +- About half of all updates were rejected. In seed 1, run 1, the update for the middle shard was rejected in every round from round 2 on, and the update for the first shard from round 6 on; from round 6 only the last shard, which sits next to the held-out tail, still helped. The score gate did its job, and that volunteer earned no credit for those rounds. +- None of this says anything about a real model. The learner is a next-byte toy, the corpus is one small file, the network is one machine's loopback, and the held-out set is scored many times (the leakage risk listed in `docs/design/volunteer-compute-v0.md` applies). diff --git a/scripts/demo-local.sh b/scripts/demo-local.sh new file mode 100755 index 0000000..3b94910 --- /dev/null +++ b/scripts/demo-local.sh @@ -0,0 +1,95 @@ +#!/usr/bin/env bash +# Volunteer compute v0 on one machine. +# +# For each seed: start oh-coordinator and three oh-volunteer processes on +# loopback for ROUNDS rounds, then run the single-process baseline +# (oh-coordinator --baseline: same corpus, same shards, same total steps, no +# compression, no averaging). Prints final held-out bits per byte for both, +# then the mean and spread over the seeds. +# +# Corpus: the repository LICENSE file (fixed text, so runs repeat). +# Settings can be changed through the environment, for example: +# ROUNDS=10 STEPS=500 DIM=32 SEEDS="1 2 3" scripts/demo-local.sh +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +ROUNDS="${ROUNDS:-10}" +STEPS="${STEPS:-500}" +DIM="${DIM:-32}" +SEEDS="${SEEDS:-1 2 3}" +BASE_PORT="${BASE_PORT:-18100}" +VOLUNTEERS=3 + +# Set LOG_DIR to keep every process log; otherwise logs go to a temporary +# directory that is removed on exit. +if [ -n "${LOG_DIR:-}" ]; then + mkdir -p "$LOG_DIR" + WORK="$(cd "$LOG_DIR" && pwd)" +else + WORK="$(mktemp -d)" +fi +cleanup() { + local pids + pids="$(jobs -p)" + if [ -n "$pids" ]; then kill $pids 2>/dev/null || true; fi + if [ -z "${LOG_DIR:-}" ]; then rm -rf "$WORK"; fi +} +trap cleanup EXIT + +echo "Building release binaries..." +cargo build --release --quiet -p oh-coordinator -p oh-volunteer --manifest-path "$ROOT/Cargo.toml" +BIN="$ROOT/target/release" + +CORPUS="$ROOT/LICENSE" +echo "Corpus: LICENSE, $(wc -c < "$CORPUS") bytes, sha256 $(sha256sum "$CORPUS" | cut -c1-16)..." +echo "Settings: rounds=$ROUNDS volunteers=$VOLUNTEERS steps=$STEPS dim=$DIM seeds=\"$SEEDS\"" +echo + +common=(--corpus "$CORPUS" --rounds "$ROUNDS" --volunteers-per-round "$VOLUNTEERS" + --steps "$STEPS" --dim "$DIM") +dist_results=() +base_results=() + +for seed in $SEEDS; do + port=$((BASE_PORT + seed)) + coord_log="$WORK/coordinator-$seed.log" + "$BIN/oh-coordinator" "${common[@]}" --seed "$seed" \ + --listen "127.0.0.1:$port" --ledger "$WORK/ledger-$seed.jsonl" > "$coord_log" 2>&1 & + coord_pid=$! + + vol_pids=() + for i in $(seq 1 "$VOLUNTEERS"); do + "$BIN/oh-volunteer" --coordinator "127.0.0.1:$port" --rounds "$ROUNDS" \ + --seed $((seed + i)) > "$WORK/volunteer-$seed-$i.log" 2>&1 & + vol_pids+=($!) + done + for pid in "${vol_pids[@]}"; do wait "$pid"; done + wait "$coord_pid" + + base_log="$WORK/baseline-$seed.log" + "$BIN/oh-coordinator" "${common[@]}" --seed "$seed" --baseline > "$base_log" 2>&1 + + start=$(grep '^start heldout_bpb=' "$coord_log" | cut -d= -f2) + dist=$(grep '^final_heldout_bpb=' "$coord_log" | cut -d= -f2) + base=$(grep '^final_heldout_bpb=' "$base_log" | cut -d= -f2) + applied=$(grep -c 'applied=yes' "$coord_log" || true) + accepted=$(cat "$WORK"/volunteer-"$seed"-*.log | grep -c 'accepted' || true) + dist_results+=("$dist") + base_results+=("$base") + + echo "Seed $seed: start $start bits/byte; volunteers $dist; baseline $base" + echo " rounds applied: $applied of $ROUNDS; updates accepted: $accepted of $((ROUNDS * VOLUNTEERS))" + echo " volunteer 1 said: $(tail -n 1 "$WORK/volunteer-$seed-1.log")" +done + +# Mean and sample standard deviation of the arguments. +stats() { + printf '%s\n' "$@" | awk '{ s += $1; q += $1 * $1; n++ } + END { m = s / n; v = (n > 1) ? (q - n * m * m) / (n - 1) : 0; if (v < 0) v = 0; + printf "mean %.4f, spread (sample std dev) %.4f, n %d", m, sqrt(v), n }' +} + +echo +echo "Final held-out bits per byte (lower is better):" +echo " $VOLUNTEERS volunteers over loopback: $(stats "${dist_results[@]}")" +echo " single-process baseline: $(stats "${base_results[@]}")"