From 457c53849d53aecb0ee0a51dca21cf712aab9318 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Mon, 9 Nov 2026 04:30:37 +0000 Subject: [PATCH] feat(train, infernet): p2p dataset/adapter pieces, multi-turn chats, native p2p chat train (infernet.text.train, `c0mpute train lora`): - The corpus goes out as content-addressed 1 MiB pieces from the buyer's chunk store; the worker pulls them over libp2p (InputRef::fetch_to_file, streamed) and checks the sha256 the offer names. Training events are published on the job topic (run_id, step events throttled to 4/s). The worker stores the adapter files as pieces behind an AdapterManifest and its receipt carries the manifest's content hash; the buyer fetches and keeps the adapter only if the weights match the manifest's sha256. Nothing needs a reachable HTTP address any more. - `c0mpute infernet text chat` (infernet.text.infer) is p2p the same way: adapter out as pieces (fetched on a worker cache miss), answer back as a chunk named by the receipt. - Compatibility: specs keep the HTTP fields optional. Workers before this release require dataset_url/adapter_url, so they never bid on a pieces-only job; `--legacy-http` makes the buyer also run the transfer server for them (receipt `sha256:` = the old HTTP path, checked as before). Old buyers' HTTP specs still parse and run on new workers. - Multi-turn chats: messages records are kept whole (every turn, in order) and prompt/completion becomes a one-exchange conversation, instead of collapsing to a "Brief me on: " record. Infernet's runner trains them with the loss on assistant turns (infernetprotocol/infernet-protocol#161) and reports formats:["messages"]; workers then advertise infernet:text:format:messages and buyers whose corpus has conversations require it. {title, text} records are unchanged. - module.toml: available. infernet (infernet.inference): - New built-in workload c0mpute-core/src/inference.rs: workers answer chat turns from their own OpenAI-compatible runtime (Ollama at 127.0.0.1:11434 by default, C0MPUTE_INFERENCE_URL for llama-server etc., `off` disables). Role `inference`, on by default, advertised only while /v1/models lists chat models (refreshed every minute): infernet:inference plus a tag per model. Answers are stored as chunks and checked against the receipt. - `c0mpute chat --p2p` (the default when not signed in) runs one auction per turn over libp2p, REPL included; the jobs bridge accepts it too. No c0mpute.com and no infernet binary on either side. - module.toml: available; temporal and text workers still need the infernet CLI for Infernet's own runtimes. Tests: text_train_e2e (3 MiB mixed corpus as pieces on loopback with a fake peft runner, gossip progress, adapter pieces verified, then adapter chat p2p), inference_e2e (multi-turn chat p2p via the bridge; also passed against the live Ollama on dev2), unit tests for specs, compat, conversion and tags. Co-Authored-By: Claude Opus 5.5 --- node/crates/c0mpute-cli/src/ai_chat.rs | 154 ++++- node/crates/c0mpute-cli/src/main.rs | 4 +- node/crates/c0mpute-cli/src/text.rs | 29 +- node/crates/c0mpute-cli/src/train.rs | 38 +- node/crates/c0mpute-core/src/capabilities.rs | 9 + node/crates/c0mpute-core/src/config.rs | 1 + node/crates/c0mpute-core/src/dispatch.rs | 7 +- node/crates/c0mpute-core/src/inference.rs | 358 ++++++++++ node/crates/c0mpute-core/src/jobs_gateway.rs | 12 +- node/crates/c0mpute-core/src/lib.rs | 1 + node/crates/c0mpute-core/src/pieces.rs | 50 ++ .../c0mpute-core/src/status_aggregator.rs | 4 +- node/crates/c0mpute-core/src/supervisor.rs | 26 +- node/crates/c0mpute-core/src/temporal/node.rs | 16 +- node/crates/c0mpute-core/src/text.rs | 616 ++++++++++++++++-- node/crates/c0mpute-core/src/train.rs | 113 +++- .../tests/fixtures/fake-infernet-text | 69 ++ .../c0mpute-core/tests/inference_e2e.rs | 146 +++++ .../c0mpute-core/tests/text_train_e2e.rs | 185 ++++++ .../crates/c0mpute-core/tests/workflow_e2e.rs | 3 +- node/crates/c0mpute-proto/src/lib.rs | 4 + plugins/infernet/module.toml | 34 +- plugins/train/module.toml | 45 +- 23 files changed, 1768 insertions(+), 156 deletions(-) create mode 100644 node/crates/c0mpute-core/src/inference.rs create mode 100755 node/crates/c0mpute-core/tests/fixtures/fake-infernet-text create mode 100644 node/crates/c0mpute-core/tests/inference_e2e.rs create mode 100644 node/crates/c0mpute-core/tests/text_train_e2e.rs diff --git a/node/crates/c0mpute-cli/src/ai_chat.rs b/node/crates/c0mpute-cli/src/ai_chat.rs index bce77603..ccc1f600 100644 --- a/node/crates/c0mpute-cli/src/ai_chat.rs +++ b/node/crates/c0mpute-cli/src/ai_chat.rs @@ -1,6 +1,10 @@ -//! `c0mpute chat`: chat with models on the c0mpute network through c0mpute.com. +//! `c0mpute chat`: chat with models on the c0mpute network, either peer to +//! peer (`--p2p`, and the default when not signed in: an `infernet.inference` +//! auction over libp2p, answered by a worker's own Ollama/llama-server; see +//! `c0mpute_core::inference`) or through c0mpute.com with a signed-in key. //! //! ```text +//! c0mpute chat --p2p "explain LoRA" # a worker answers over libp2p, no account //! c0mpute chat login # device sign-in: approve on c0mpute.com with CoinPay (DID required) //! c0mpute chat "explain LoRA" # one answer, streamed //! c0mpute chat # interactive session (TTY) @@ -52,6 +56,16 @@ struct Args { /// Print the full JSON response instead of streaming text. #[arg(long)] json: bool, + /// Ask a worker directly over libp2p (`infernet.inference`): no sign-in, + /// no c0mpute.com. The default when this machine is not signed in. + #[arg(long)] + p2p: bool, + /// With --p2p: the most you pay a worker for one answer, in USD. + #[arg(long, default_value_t = 0.01)] + max_price: f64, + /// With --p2p: seconds to wait for an answer. + #[arg(long, default_value_t = 300)] + timeout_s: u64, /// c0mpute site URL (default https://c0mpute.com, or $C0MPUTE_URL). #[arg(long)] url: Option, @@ -151,7 +165,10 @@ pub async fn run(args: &[String]) -> Result<()> { let a = Args::try_parse_from(std::iter::once("c0mpute chat".to_string()).chain(args.iter().cloned())) .unwrap_or_else(|e| e.exit()); let url = base_url(a.url.as_deref()); - let key = require_key()?; + let p2p = a.p2p || api_key().is_none(); + if p2p && !a.p2p { + eprintln!("(not signed in: asking a c0mpute worker peer to peer; `c0mpute chat login` uses c0mpute.com instead)"); + } let mut prompt = a.prompt.join(" "); let stdin = std::io::stdin(); @@ -171,9 +188,23 @@ pub async fn run(args: &[String]) -> Result<()> { if !std::io::stdin().is_terminal() { bail!("no prompt: pass it as an argument or pipe it on stdin"); } - return repl(&url, &key, &a, messages).await; + if p2p { + return P2pChat::boot(&a).await?.repl(&a, messages).await; + } + return repl(&url, &require_key()?, &a, messages).await; } messages.push(json!({"role": "user", "content": prompt})); + if p2p { + let chat = P2pChat::boot(&a).await?; + let (text, raw) = chat.ask(&a, a.model.clone(), &messages).await?; + if a.json { + println!("{}", serde_json::to_string_pretty(&raw)?); + } else { + println!("{text}"); + } + return Ok(()); + } + let key = require_key()?; let answer = complete(&url, &key, &a, &messages, a.json).await?; if a.json { println!("{answer}"); @@ -300,11 +331,107 @@ fn clone_args(a: &Args) -> Args { max_tokens: a.max_tokens, temperature: a.temperature, json: false, + p2p: a.p2p, + max_price: a.max_price, + timeout_s: a.timeout_s, url: a.url.clone(), prompt: vec![], } } +/// The `infernet.inference` spec for one turn. +fn p2p_spec(a: &Args, model: Option, messages: &[Value]) -> Result { + let mut spec = json!({"v": 1, "messages": messages}); + if let Some(m) = model { + spec["model"] = json!(m); + } + if let Some(t) = a.max_tokens { + spec["max_tokens"] = json!(t); + } + if let Some(t) = a.temperature { + spec["temperature"] = json!(t); + } + c0mpute_core::inference::parse_spec(&spec).map_err(|e| anyhow!(e))?; + Ok(spec) +} + +/// Chat with workers over libp2p: one buyer node for the whole session, one +/// `infernet.inference` auction per turn, answers fetched from the winner and +/// checked against its receipt. +struct P2pChat { + bridge: c0mpute_core::jobs_gateway::JobsBridge, +} + +impl P2pChat { + async fn boot(a: &Args) -> Result { + let sup = c0mpute_core::Supervisor::boot_buyer(c0mpute_core::Config::default()).await?; + // Any token: the bridge's HTTP side is never mounted here. + let token: String = std::iter::repeat_n('x', 32).collect(); + let bridge = c0mpute_core::jobs_gateway::JobsBridge::new(sup.libp2p, sup.store, token, a.max_price, Vec::new())?; + // Let the swarm reach peers before the first auction. + tokio::time::sleep(Duration::from_secs(5)).await; + Ok(Self { bridge }) + } + + async fn ask(&self, a: &Args, model: Option, messages: &[Value]) -> Result<(String, Value)> { + let spec = p2p_spec(a, model, messages)?; + let r = self + .bridge + .submit_once(c0mpute_core::inference::WORKLOAD, spec, Duration::from_secs(a.timeout_s.clamp(10, 900))) + .await + .context("no worker answered (none serving infernet.inference for this model, or all over --max-price)")?; + let v = r.get("result").cloned().unwrap_or(r.clone()); + if v["ok"] != json!(true) { + bail!("the worker could not answer: {}", v["error"].as_str().unwrap_or("?")); + } + Ok((v["text"].as_str().unwrap_or_default().to_string(), r)) + } + + async fn repl(&self, a: &Args, mut messages: Vec) -> Result<()> { + let mut model = a.model.clone(); + eprintln!("c0mpute chat (peer to peer) · {} · /model , /reset, /exit", model.as_deref().unwrap_or("any served model")); + let stdin = std::io::stdin(); + let base = messages.clone(); + loop { + eprint!("\x1b[32m> \x1b[0m"); + std::io::stderr().flush().ok(); + let mut line = String::new(); + if stdin.lock().read_line(&mut line)? == 0 { + eprintln!(); + return Ok(()); + } + let line = line.trim(); + match line { + "" => continue, + "/exit" | "/quit" => return Ok(()), + "/reset" => { + messages = base.clone(); + eprintln!("(new conversation)"); + continue; + } + _ if line.starts_with("/model") => { + let m = line.trim_start_matches("/model").trim(); + model = (!m.is_empty() && m != "auto").then(|| m.to_string()); + eprintln!("(model: {})", model.as_deref().unwrap_or("any served model")); + continue; + } + _ => {} + } + messages.push(json!({"role": "user", "content": line})); + match self.ask(a, model.clone(), &messages).await { + Ok((answer, _)) => { + println!("{answer}"); + messages.push(json!({"role": "assistant", "content": answer})); + } + Err(e) => { + messages.pop(); + eprintln!("error: {e:#}"); + } + } + } + } +} + async fn login(rest: &[String]) -> Result<()> { let mut url_flag = None; let mut it = rest.iter(); @@ -497,6 +624,27 @@ mod tests { assert_eq!(a.model.as_deref(), Some("qwen2.5:7b")); assert_eq!(a.system.as_deref(), Some("terse")); assert_eq!(a.prompt.join(" "), "what is 2+2"); + assert!(!a.p2p); + } + + #[test] + fn p2p_turn_is_a_valid_inference_spec() { + let a = Args::try_parse_from(["c0mpute chat", "--p2p", "--max-tokens", "64", "--temperature", "0.3", "hi"]).unwrap(); + assert!(a.p2p); + let msgs = vec![json!({"role": "system", "content": "terse"}), json!({"role": "user", "content": "hi"})]; + let spec = p2p_spec(&a, Some("qwen2.5:0.5b".into()), &msgs).unwrap(); + assert_eq!(spec["v"], 1); + assert_eq!(spec["model"], "qwen2.5:0.5b"); + assert_eq!(spec["max_tokens"], 64); + assert_eq!( + c0mpute_core::inference::required_capabilities(&spec), + vec!["infernet:inference".to_string(), "infernet:inference:model:qwen2.5:0.5b".to_string()] + ); + let any = p2p_spec(&a, None, &msgs).unwrap(); + assert!(any.get("model").is_none()); + // A turn the workers would refuse never leaves this machine. + let bad = Args::try_parse_from(["c0mpute chat", "--p2p", "--temperature", "9", "hi"]).unwrap(); + assert!(p2p_spec(&bad, None, &msgs).is_err()); } #[test] diff --git a/node/crates/c0mpute-cli/src/main.rs b/node/crates/c0mpute-cli/src/main.rs index 0395e71f..c06937a1 100644 --- a/node/crates/c0mpute-cli/src/main.rs +++ b/node/crates/c0mpute-cli/src/main.rs @@ -47,7 +47,7 @@ use clap::{Parser, Subcommand}; name = "c0mpute", version, about = "c0mpute.com — decentralized compute network", - long_about = "c0mpute.com CLI. Submit jobs, run a worker, manage modules.\n\nBuilt-in workloads:\n transcode (FFmpeg)\n tts (Piper, Kokoro, or Pocket, CPU)\n stt (faster-whisper, CPU)\n workflow-step (JMESPath transforms + guarded HTTP calls, for meshhook)\n ocr (Tesseract on the worker, images + PDFs)\n embed (fastembed bge-small-en-v1.5, CPU)\n render (Blender frames, Cycles on CPU or GPU)\n coinpay (DID + payments, peer CLI)\n infernet (AI inference, peer CLI)\n\n c0mpute coinpay reputation did claim\n c0mpute transcode preset list\n c0mpute tts setup --engine all\n c0mpute stt setup\n c0mpute speech serve\n c0mpute ocr scan.pdf --lang eng\n c0mpute embed \"first text\" \"second text\"\n c0mpute render scene.blend --frames 1-120 --out frames --mp4\n C0MPUTE_JOBS_TOKEN= c0mpute jobs serve\n c0mpute worker start --roles tts,stt\n c0mpute chat login\n c0mpute chat \"hello\"\n c0mpute infernet chat --model qwen2.5:7b \"hello\"\n c0mpute infernet model list --node \n c0mpute infernet reservation pricing\n c0mpute infernet text serve\n c0mpute infernet text train --job job.json --dataset corpus.jsonl --yes\n c0mpute infernet text chat --adapter ./adapter \"How should I price my SaaS?\"\n c0mpute train lora --base Qwen/Qwen2.5-0.5B-Instruct --dataset data.jsonl --yes\n c0mpute gameserver start minecraft --hours 4 --ram 4g --accept-eula\n c0mpute infernet temporal serve --model sha256: --train\n c0mpute infernet temporal predict --session s1 --input step.json\n c0mpute webhook serve --price 0.001\n c0mpute webhook send https://example.com/hook -d '{\"event\":\"paid\"}' --secret whsec_..." + long_about = "c0mpute.com CLI. Submit jobs, run a worker, manage modules.\n\nBuilt-in workloads:\n transcode (FFmpeg)\n tts (Piper, Kokoro, or Pocket, CPU)\n stt (faster-whisper, CPU)\n workflow-step (JMESPath transforms + guarded HTTP calls, for meshhook)\n ocr (Tesseract on the worker, images + PDFs)\n embed (fastembed bge-small-en-v1.5, CPU)\n render (Blender frames, Cycles on CPU or GPU)\n coinpay (DID + payments, peer CLI)\n inference (chat from your Ollama/llama-server, p2p)\n infernet (temporal + LoRA runtimes, peer CLI)\n\n c0mpute coinpay reputation did claim\n c0mpute transcode preset list\n c0mpute tts setup --engine all\n c0mpute stt setup\n c0mpute speech serve\n c0mpute ocr scan.pdf --lang eng\n c0mpute embed \"first text\" \"second text\"\n c0mpute render scene.blend --frames 1-120 --out frames --mp4\n C0MPUTE_JOBS_TOKEN= c0mpute jobs serve\n c0mpute worker start --roles tts,stt\n c0mpute chat --p2p \"hello\" (peer to peer, no account)\n c0mpute chat login\n c0mpute chat \"hello\"\n c0mpute infernet chat --model qwen2.5:7b \"hello\"\n c0mpute infernet model list --node \n c0mpute infernet reservation pricing\n c0mpute infernet text serve\n c0mpute infernet text train --job job.json --dataset corpus.jsonl --yes\n c0mpute infernet text chat --adapter ./adapter \"How should I price my SaaS?\"\n c0mpute train lora --base Qwen/Qwen2.5-0.5B-Instruct --dataset data.jsonl --yes\n c0mpute gameserver start minecraft --hours 4 --ram 4g --accept-eula\n c0mpute infernet temporal serve --model sha256: --train\n c0mpute infernet temporal predict --session s1 --input step.json\n c0mpute webhook serve --price 0.001\n c0mpute webhook send https://example.com/hook -d '{\"event\":\"paid\"}' --secret whsec_..." )] struct Cli { /// Override the config file location. @@ -1553,6 +1553,7 @@ async fn run_worker(cmd: WorkerCmd, config_path: &std::path::Path) -> Result<()> Role::Embed, Role::LiveStream, Role::Render, + Role::Inference, ] { if !cfg.roles.contains(&role) { cfg.roles.push(role); @@ -3897,6 +3898,7 @@ fn parse_role(s: &str) -> Option { "gameserver" | "gameservers" | "games" => Some(Role::Gameserver), "live-stream" | "live_stream" | "livestream" | "live" | "stream" => Some(Role::LiveStream), "render" | "blender" => Some(Role::Render), + "inference" | "chat" | "llm" => Some(Role::Inference), other => { eprintln!("warning: unknown role '{other}' ignored"); None diff --git a/node/crates/c0mpute-cli/src/text.rs b/node/crates/c0mpute-cli/src/text.rs index ef0badf2..1e1c9f56 100644 --- a/node/crates/c0mpute-cli/src/text.rs +++ b/node/crates/c0mpute-cli/src/text.rs @@ -57,7 +57,10 @@ enum Sub { worker: Vec, #[arg(long, default_value_t = 0.05)] max_price: f64, - /// How the worker reaches this machine's transfer server (use https off-LAN). + /// Also serve workers older than the peer-to-peer transport over HTTP from this machine. + #[arg(long)] + legacy_http: bool, + /// With --legacy-http: how an old worker reaches this machine's transfer server (use https off-LAN). #[arg(long)] public_url: Option, #[arg(long, default_value = "0.0.0.0:0")] @@ -91,7 +94,10 @@ enum Sub { gpu: bool, #[arg(long)] min_ram_gb: Option, - /// How the worker reaches this machine's transfer server (use https off-LAN). + /// Also serve workers older than the peer-to-peer transport over HTTP from this machine. + #[arg(long)] + legacy_http: bool, + /// With --legacy-http: how an old worker reaches this machine's transfer server (use https off-LAN). #[arg(long)] public_url: Option, #[arg(long, default_value = "0.0.0.0:0")] @@ -148,7 +154,7 @@ pub async fn run(args: &[String]) -> Result<()> { tokio::signal::ctrl_c().await?; Ok(()) } - Sub::Chat { adapter, question, no_search, search_url, max_new_tokens, worker, max_price, public_url, bind, timeout_s, bid_window_s, json: as_json, net } => { + Sub::Chat { adapter, question, no_search, search_url, max_new_tokens, worker, max_price, legacy_http, public_url, bind, timeout_s, bid_window_s, json: as_json, net } => { let question = question.join(" "); anyhow::ensure!(!question.trim().is_empty(), "ask a question: c0mpute infernet text chat --adapter ./adapter \"How should I price my SaaS?\""); let req = text::ChatRequest { @@ -158,15 +164,18 @@ pub async fn run(args: &[String]) -> Result<()> { max_new_tokens, workers: worker, max_price_usd: max_price, + legacy_http, public_url, bind, timeout_s, bid_window: Duration::from_secs(bid_window_s), }; - let n = c0mpute_core::temporal::node::spawn_node(&net_opts(&net, "client")).await?; + // The adapter goes out as pieces from this store, served over libp2p. + let store = c0mpute_store::ChunkStore::open(data_dir().join("store")).await?; + let n = c0mpute_core::temporal::node::spawn_node_with_store(&net_opts(&net, "client"), store.clone()).await?; tokio::time::sleep(Duration::from_secs(2)).await; eprintln!("asking a c0mpute worker ..."); - let out = text::chat(n, &req).await?; + let out = text::chat(n, &store, &req).await?; if as_json { println!("{}", serde_json::to_string(&out)?); } else { @@ -178,8 +187,8 @@ pub async fn run(args: &[String]) -> Result<()> { } Ok(()) } - Sub::Train { job, dataset, output, worker, max_price, gpu, min_ram_gb, public_url, bind, timeout_s, retries, bid_window_s, yes, json: as_json, net } => { - let req = TrainRequest { job_path: job, dataset_path: dataset, output, workers: worker, max_price_usd: max_price, gpu, min_ram_gb, public_url, bind, timeout_s, retries, bid_window: Duration::from_secs(bid_window_s) }; + Sub::Train { job, dataset, output, worker, max_price, gpu, min_ram_gb, legacy_http, public_url, bind, timeout_s, retries, bid_window_s, yes, json: as_json, net } => { + let req = TrainRequest { job_path: job, dataset_path: dataset, output, workers: worker, max_price_usd: max_price, gpu, min_ram_gb, legacy_http, public_url, bind, timeout_s, retries, bid_window: Duration::from_secs(bid_window_s) }; let (_, _, p) = text::load(&req)?; if !yes { eprintln!( @@ -198,12 +207,14 @@ pub async fn run(args: &[String]) -> Result<()> { } std::process::exit(3); } - let n = c0mpute_core::temporal::node::spawn_node(&net_opts(&net, "client")).await?; + // The corpus goes out as pieces from this store, served over libp2p. + let store = c0mpute_store::ChunkStore::open(data_dir().join("store")).await?; + let n = c0mpute_core::temporal::node::spawn_node_with_store(&net_opts(&net, "client"), store.clone()).await?; tokio::time::sleep(Duration::from_secs(2)).await; eprintln!("looking for a worker for {} ...", p.name); let mut total = None; let mut on_step = false; - let out = text::train(n, &req, |ev| { + let out = text::train(n, &store, &req, |ev| { if as_json { println!("{ev}"); return; diff --git a/node/crates/c0mpute-cli/src/train.rs b/node/crates/c0mpute-cli/src/train.rs index d3f132d4..2e0be198 100644 --- a/node/crates/c0mpute-cli/src/train.rs +++ b/node/crates/c0mpute-cli/src/train.rs @@ -8,7 +8,8 @@ use std::path::PathBuf; use std::time::Duration; use anyhow::{Context, Result}; -use c0mpute_core::temporal::node::{NetOptions, spawn_node}; +use c0mpute_core::temporal::node::{NetOptions, spawn_node_with_store}; +use c0mpute_store::ChunkStore; use c0mpute_core::text::{self, TrainRequest}; use c0mpute_core::train::{LoraOptions, normalize_dataset}; use clap::{Args, Parser, Subcommand}; @@ -39,8 +40,10 @@ enum Sub { /// Train a LoRA adapter on a remote worker and download it. /// /// The dataset is JSONL: {"text"} (optionally "title"), {"prompt","completion"} - /// or {"messages":[{role,content}]}. Workers run `infernet model train` - /// (peft, fp32 LoRA on CPU or CUDA); keep the base model small. + /// or {"messages":[{role,content}]} (multi-turn chats train on every turn, + /// loss on the assistant's). Workers run `infernet model train` (peft, fp32 + /// LoRA on CPU or CUDA); keep the base model small. The dataset and the + /// adapter move peer to peer over libp2p. Lora { /// Base model, a Hugging Face id (e.g. Qwen/Qwen2.5-0.5B-Instruct). #[arg(long)] @@ -81,7 +84,11 @@ enum Sub { gpu: bool, #[arg(long)] min_ram_gb: Option, - /// How the worker reaches this machine's transfer server (needed off-LAN: an https URL that forwards to --bind). + /// Also serve workers older than the peer-to-peer transport: the corpus and adapter then go + /// over HTTP from this machine (the worker must reach it: same LAN, or --public-url). + #[arg(long)] + legacy_http: bool, + /// With --legacy-http: how an old worker reaches this machine's transfer server (an https URL forwarding to --bind). #[arg(long)] public_url: Option, #[arg(long, default_value = "0.0.0.0:0")] @@ -125,6 +132,7 @@ pub async fn run(args: &[String]) -> Result<()> { max_price, gpu, min_ram_gb, + legacy_http, public_url, bind, timeout_s, @@ -161,6 +169,7 @@ pub async fn run(args: &[String]) -> Result<()> { max_price_usd: max_price, gpu, min_ram_gb, + legacy_http, public_url, bind, timeout_s, @@ -189,18 +198,23 @@ pub async fn run(args: &[String]) -> Result<()> { std::process::exit(3); } - let node = spawn_node(&NetOptions { - identity_dir: data_dir().join("client"), - listen: net.listen.clone(), - bootstrap: net.bootstrap.clone(), - default_bootstrap: !net.no_default_bootstrap, - }) + // The corpus goes out as pieces from this store, served over libp2p. + let store = ChunkStore::open(data_dir().join("store")).await?; + let node = spawn_node_with_store( + &NetOptions { + identity_dir: data_dir().join("client"), + listen: net.listen.clone(), + bootstrap: net.bootstrap.clone(), + default_bootstrap: !net.no_default_bootstrap, + }, + store.clone(), + ) .await?; tokio::time::sleep(Duration::from_secs(2)).await; - eprintln!("looking for a worker advertising infernet:text:train ..."); + eprintln!("looking for a worker advertising {} ...", text::required_tags(&corpus).join(" + ")); let mut total = None; let mut on_step = false; - let result = text::train(node, &req, |ev| { + let result = text::train(node, &store, &req, |ev| { if as_json { println!("{ev}"); return; diff --git a/node/crates/c0mpute-core/src/capabilities.rs b/node/crates/c0mpute-core/src/capabilities.rs index b5fdc697..b941db39 100644 --- a/node/crates/c0mpute-core/src/capabilities.rs +++ b/node/crates/c0mpute-core/src/capabilities.rs @@ -109,6 +109,15 @@ pub fn role_tags(config: &Config) -> Vec { } } Role::LiveStream => tags.push(crate::live::ROLE_TAG.to_string()), + Role::Inference => { + // Only while the local runtime lists chat models (refreshed + // every minute), one tag per model. + let models = crate::inference::ready_models(); + if !models.is_empty() { + tags.push(crate::inference::ROLE_TAG.to_string()); + tags.extend(models.iter().map(|m| crate::inference::model_tag(m))); + } + } Role::Render => { // Only where `blender` runs. The probe (a few seconds) runs // once at worker start; until it finishes, nothing is shown. diff --git a/node/crates/c0mpute-core/src/config.rs b/node/crates/c0mpute-core/src/config.rs index 03813976..71eb139d 100644 --- a/node/crates/c0mpute-core/src/config.rs +++ b/node/crates/c0mpute-core/src/config.rs @@ -314,6 +314,7 @@ impl Default for Config { Role::Embed, Role::LiveStream, Role::Render, + Role::Inference, ], update_channel: default_update_channel(), update_auto: default_auto_update(), diff --git a/node/crates/c0mpute-core/src/dispatch.rs b/node/crates/c0mpute-core/src/dispatch.rs index a0171a3d..6b3909ad 100644 --- a/node/crates/c0mpute-core/src/dispatch.rs +++ b/node/crates/c0mpute-core/src/dispatch.rs @@ -519,8 +519,8 @@ async fn handle_offer( /// Workload types this worker should listen for, derived from its /// configured roles. Each modality maps to a separate job topic. -/// `infernet.inference` is handled by the infernet peer binary, not -/// this daemon. +/// `infernet.inference` is served natively from the box's own +/// OpenAI-compatible runtime ([`crate::inference`]). pub fn workload_types_from_roles(roles: &[c0mpute_proto::Role]) -> Vec { let mut types = Vec::new(); if roles.contains(&c0mpute_proto::Role::Transcode) { @@ -555,6 +555,9 @@ pub fn workload_types_from_roles(roles: &[c0mpute_proto::Role]) -> Vec { if roles.contains(&c0mpute_proto::Role::Embed) { types.push(crate::embed::WORKLOAD.to_string()); } + if roles.contains(&c0mpute_proto::Role::Inference) { + types.push(crate::inference::WORKLOAD.to_string()); + } if roles.contains(&c0mpute_proto::Role::Render) { types.push(crate::render::WORKLOAD.to_string()); } diff --git a/node/crates/c0mpute-core/src/inference.rs b/node/crates/c0mpute-core/src/inference.rs new file mode 100644 index 00000000..dea76cab --- /dev/null +++ b/node/crates/c0mpute-core/src/inference.rs @@ -0,0 +1,358 @@ +//! `infernet.inference`: chat completions from open models, peer to peer, +//! served by the worker's own OpenAI-compatible runtime. +//! +//! A buyer (`c0mpute chat --p2p`, or the `c0mpute jobs serve` bridge) auctions +//! a spec of chat messages on `c0mpute/jobs/infernet.inference`. A worker +//! answers it from the local runtime it already runs (Ollama on +//! `127.0.0.1:11434` by default, or any OpenAI-compatible server such as +//! llama.cpp's `llama-server`, set with `C0MPUTE_INFERENCE_URL`), stores the +//! answer as a chunk and publishes a receipt with its hash; the buyer fetches +//! it from the worker and checks it. No c0mpute.com, no `infernet` binary on +//! either side. +//! +//! Workers advertise `infernet:inference` plus one +//! `infernet:inference:model:` tag per model the runtime lists at +//! `/v1/models`, and only while it answers; the list is refreshed every +//! minute. The runtime URL is operator configuration, never from a spec, so a +//! buyer cannot point a worker at another address. `C0MPUTE_INFERENCE_URL=off` +//! turns the role off. +//! +//! Answers are one-shot (no token streaming across the network yet). + +use std::sync::{Arc, RwLock}; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result}; +use c0mpute_net::topics::JobStatus; +use c0mpute_net::{Libp2pNetwork, Network}; +use c0mpute_store::ChunkStore; +use serde::Deserialize; +use serde_json::{Value, json}; +use tokio::sync::mpsc; +use tracing::{info, warn}; + +use crate::runner::{AcceptedJob, WorkerLoad, publish_receipt}; + +pub const WORKLOAD: &str = "infernet.inference"; +/// Capability tag: this node answers chat completions. +pub const ROLE_TAG: &str = "infernet:inference"; +pub const DEFAULT_RUNTIME_URL: &str = "http://127.0.0.1:11434"; +pub const RUNTIME_URL_ENV: &str = "C0MPUTE_INFERENCE_URL"; +/// Concurrent answers per node: the local runtime is usually one GPU or a CPU. +pub const MAX_CONCURRENT: usize = 2; +pub const MAX_MESSAGES: usize = 64; +pub const MAX_PROMPT_BYTES: usize = 64 * 1024; +pub const MAX_TOKENS: u32 = 4096; +pub const DEFAULT_MAX_TOKENS: u32 = 512; +pub const MAX_RESULT_BYTES: usize = 256 * 1024; +const GENERATION_TIMEOUT: Duration = Duration::from_secs(300); +const REFRESH_EVERY: Duration = Duration::from_secs(60); + +pub fn model_tag(model: &str) -> String { + format!("{ROLE_TAG}:model:{model}") +} + +/// What a buyer asks for. +#[derive(Debug, Clone, Deserialize, PartialEq)] +pub struct ChatSpec { + pub v: u32, + /// A model id the worker's runtime lists; None means any. + #[serde(default)] + pub model: Option, + pub messages: Vec, + #[serde(default)] + pub max_tokens: Option, + #[serde(default)] + pub temperature: Option, +} + +fn valid_model_id(m: &str) -> bool { + !m.is_empty() && m.len() <= 128 && m.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.' | ':' | '/' | '@')) +} + +pub fn parse_spec(spec: &Value) -> std::result::Result { + let s: ChatSpec = serde_json::from_value(spec.clone()).map_err(|e| format!("invalid {WORKLOAD} spec: {e}"))?; + if s.v != 1 { + return Err(format!("unsupported {WORKLOAD} spec version {}", s.v)); + } + if s.model.as_deref().is_some_and(|m| !valid_model_id(m)) { + return Err("model must be 1-128 characters of letters, digits and -_.:/@".into()); + } + if s.messages.is_empty() || s.messages.len() > MAX_MESSAGES { + return Err(format!("messages must hold 1..={MAX_MESSAGES} entries")); + } + let mut bytes = 0; + for m in &s.messages { + let role = m.get("role").and_then(Value::as_str).unwrap_or(""); + if !["system", "user", "assistant"].contains(&role) { + return Err("message role must be system, user or assistant".into()); + } + bytes += m.get("content").and_then(Value::as_str).ok_or("message content must be a string")?.len(); + } + if bytes > MAX_PROMPT_BYTES { + return Err(format!("messages exceed {MAX_PROMPT_BYTES} bytes")); + } + if s.messages.last().and_then(|m| m.get("role")).and_then(Value::as_str) == Some("assistant") { + return Err("the last message must be the user's (or system's), not the assistant's".into()); + } + if s.max_tokens.is_some_and(|t| !(1..=MAX_TOKENS).contains(&t)) { + return Err(format!("max_tokens must be 1..={MAX_TOKENS}")); + } + if s.temperature.is_some_and(|t| !(0.0..=2.0).contains(&t)) { + return Err("temperature must be 0..=2".into()); + } + Ok(s) +} + +/// Tags a worker must carry for `spec`. +pub fn required_capabilities(spec: &Value) -> Vec { + let mut t = vec![ROLE_TAG.to_string()]; + if let Ok(ChatSpec { model: Some(m), .. }) = parse_spec(spec) { + t.push(model_tag(&m)); + } + t +} + +/// A result the buyer accepts: `{v:1, ok:true, text, model}` or `{v:1, ok:false, error}`. +pub fn valid_result(value: &Value) -> bool { + value.get("v").and_then(Value::as_u64) == Some(1) + && match value.get("ok").and_then(Value::as_bool) { + Some(true) => value.get("text").is_some_and(Value::is_string) && value.get("model").is_some_and(Value::is_string), + Some(false) => value.get("error").is_some_and(Value::is_string), + None => false, + } +} + +// ─── the local runtime ──────────────────────────────────────────────────── + +/// The operator's runtime base URL, or None when the role is turned off. +pub fn runtime_url() -> Option { + match std::env::var(RUNTIME_URL_ENV) { + Ok(v) if matches!(v.trim().to_ascii_lowercase().as_str(), "off" | "0" | "false" | "no" | "") => None, + Ok(v) => Some(v.trim().trim_end_matches('/').to_string()), + Err(_) => Some(DEFAULT_RUNTIME_URL.to_string()), + } +} + +static READY: RwLock> = RwLock::new(Vec::new()); + +/// Models the runtime listed at the last probe (empty: not ready, advertise nothing). +pub fn ready_models() -> Vec { + READY.read().map(|m| m.clone()).unwrap_or_default() +} + +fn set_ready(models: Vec) { + if let Ok(mut m) = READY.write() { + *m = models; + } +} + +/// Model ids at `{base}/v1/models` (OpenAI shape, as Ollama and llama-server serve it). +pub async fn list_models(base: &str) -> Result> { + let c = reqwest::Client::builder().timeout(Duration::from_secs(5)).build()?; + let v: Value = c.get(format!("{base}/v1/models")).send().await?.error_for_status()?.json().await?; + let mut models: Vec = v["data"] + .as_array() + .into_iter() + .flatten() + .filter_map(|m| m["id"].as_str()) + .filter(|m| valid_model_id(m)) + // Embedding-only models cannot chat. + .filter(|m| !m.contains("embed")) + .map(String::from) + .collect(); + models.sort(); + models.dedup(); + Ok(models) +} + +/// Probe once now, then every minute, keeping [`ready_models`] current. +pub async fn refresh_loop(base: String) { + let mut last: Option> = None; + loop { + let models = list_models(&base).await.unwrap_or_default(); + if last.as_ref() != Some(&models) { + if models.is_empty() { + info!(runtime = %base, "inference: no OpenAI-compatible runtime with chat models; not advertising {WORKLOAD}"); + } else { + info!(runtime = %base, ?models, "inference: serving {WORKLOAD}"); + } + last = Some(models.clone()); + } + set_ready(models); + tokio::time::sleep(REFRESH_EVERY).await; + } +} + +/// Answer one spec from the runtime at `base`. Never errors: a failure is an +/// `ok: false` result the buyer can read. +pub async fn execute(base: &str, spec: &Value, ready: &[String]) -> Value { + let started = Instant::now(); + let blob = match run(base, spec, ready).await { + Ok(v) => v, + Err(e) => json!({"v": 1, "ok": false, "error": format!("{e:#}")}), + }; + let mut blob = blob; + blob["duration_ms"] = json!(started.elapsed().as_millis() as u64); + if serde_json::to_vec(&blob).map(|b| b.len()).unwrap_or(usize::MAX) > MAX_RESULT_BYTES { + return json!({"v": 1, "ok": false, "error": format!("answer is over {MAX_RESULT_BYTES} bytes")}); + } + blob +} + +async fn run(base: &str, spec: &Value, ready: &[String]) -> Result { + let s = parse_spec(spec).map_err(anyhow::Error::msg)?; + let model = match &s.model { + Some(m) if ready.iter().any(|r| r == m) => m.clone(), + Some(m) => anyhow::bail!("model {m} is not served here"), + None => ready.first().cloned().context("no model is served here")?, + }; + let mut body = json!({ + "model": model, + "messages": s.messages, + "max_tokens": s.max_tokens.unwrap_or(DEFAULT_MAX_TOKENS), + "stream": false, + }); + if let Some(t) = s.temperature { + body["temperature"] = json!(t); + } + let c = reqwest::Client::builder().timeout(GENERATION_TIMEOUT).build()?; + let resp = c.post(format!("{base}/v1/chat/completions")).json(&body).send().await.context("reach the local runtime")?; + let status = resp.status(); + let v: Value = resp.json().await.context("runtime answer is not JSON")?; + if !status.is_success() { + anyhow::bail!("runtime answered {status}: {}", v["error"]["message"].as_str().or(v["error"].as_str()).unwrap_or("?")); + } + let choice = &v["choices"][0]; + let text = choice["message"]["content"].as_str().context("runtime answer has no choices[0].message.content")?; + Ok(json!({ + "v": 1, "ok": true, "model": model, "text": text, + "finish_reason": choice["finish_reason"], "usage": v["usage"], + })) +} + +// ─── worker runner ──────────────────────────────────────────────────────── + +pub fn worker_load() -> WorkerLoad { + WorkerLoad::new(Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT)), MAX_CONCURRENT) +} + +/// Bid only on valid specs for a model served right now. +pub fn offer_filter() -> crate::dispatch::OfferFilter { + Arc::new(|offer| { + let ready = ready_models(); + !ready.is_empty() + && offer.spec_inline.as_ref().and_then(|v| parse_spec(v).ok()).is_some_and(|s| s.model.is_none_or(|m| ready.contains(&m))) + }) +} + +pub fn spawn_inference_runner(net: Arc, store: ChunkStore, base: String, load: WorkerLoad) -> mpsc::Sender { + let (tx, mut rx) = mpsc::channel::(MAX_CONCURRENT * 4); + tokio::spawn(async move { + while let Some(job) = rx.recv().await { + let Ok(slot) = load.compute_slots().acquire_owned().await else { return }; + let (net, store, base, load) = (net.clone(), store.clone(), base.clone(), load.clone()); + tokio::spawn(async move { + let _slot = slot; + let started = Instant::now(); + let job_id = job.accept.job_id.clone(); + let result: Result = async { + let spec = crate::workflow::resolve_spec(&net, &job).await?; + let answer = execute(&base, &spec, &ready_models()).await; + let bytes = serde_json::to_vec(&answer)?; + let hash = store.put(&bytes).await.context("store answer")?; + net.announce(&hash).await.context("announce answer")?; + Ok(hash.to_hex()) + } + .await; + match result { + Ok(hash) => { + load.record_job(started.elapsed()); + info!(%job_id, %hash, "inference: answered"); + publish_receipt(&net, &job.accept, Some(hash), JobStatus::Completed).await; + } + Err(e) => { + warn!(%job_id, err = %format!("{e:#}"), "inference: failed"); + publish_receipt(&net, &job.accept, None, JobStatus::Failed).await; + } + } + }); + } + }); + tx +} + +#[cfg(test)] +mod tests { + use super::*; + + fn spec(v: Value) -> std::result::Result { + parse_spec(&v) + } + + #[test] + fn spec_validation() { + let ok = json!({"v": 1, "messages": [{"role": "system", "content": "be brief"}, {"role": "user", "content": "hi"}]}); + assert!(spec(ok.clone()).is_ok()); + assert_eq!(required_capabilities(&ok), vec![ROLE_TAG.to_string()]); + let m = json!({"v": 1, "model": "qwen2.5:0.5b", "messages": [{"role": "user", "content": "hi"}], "max_tokens": 64, "temperature": 0.2}); + assert!(spec(m.clone()).is_ok()); + assert_eq!(required_capabilities(&m), vec![ROLE_TAG.to_string(), "infernet:inference:model:qwen2.5:0.5b".to_string()]); + assert!(spec(json!({"v": 2, "messages": [{"role": "user", "content": "hi"}]})).unwrap_err().contains("version")); + assert!(spec(json!({"v": 1, "messages": []})).is_err()); + assert!(spec(json!({"v": 1, "messages": [{"role": "tool", "content": "x"}]})).is_err()); + assert!(spec(json!({"v": 1, "messages": [{"role": "user", "content": 3}]})).is_err()); + assert!(spec(json!({"v": 1, "messages": [{"role": "user", "content": "x".repeat(MAX_PROMPT_BYTES + 1)}]})).is_err()); + assert!(spec(json!({"v": 1, "messages": [{"role": "user", "content": "q"}, {"role": "assistant", "content": "a"}]})).unwrap_err().contains("last message")); + assert!(spec(json!({"v": 1, "model": "../etc passwd", "messages": [{"role": "user", "content": "q"}]})).is_err()); + assert!(spec(json!({"v": 1, "max_tokens": 0, "messages": [{"role": "user", "content": "q"}]})).is_err()); + assert!(spec(json!({"v": 1, "temperature": 3.0, "messages": [{"role": "user", "content": "q"}]})).is_err()); + } + + #[test] + fn results() { + assert!(valid_result(&json!({"v": 1, "ok": true, "text": "hi", "model": "m"}))); + assert!(valid_result(&json!({"v": 1, "ok": false, "error": "no"}))); + assert!(!valid_result(&json!({"v": 1, "ok": true}))); + assert!(!valid_result(&json!({"v": 2, "ok": false, "error": "no"}))); + } + + /// A stand-in OpenAI-compatible runtime on loopback. + pub(crate) async fn fake_runtime() -> String { + use axum::{Json, Router, routing::{get, post}}; + let app = Router::new() + .route("/v1/models", get(|| async { Json(json!({"object": "list", "data": [{"id": "tiny:1b"}, {"id": "nomic-embed-text"}]})) })) + .route( + "/v1/chat/completions", + post(|Json(b): Json| async move { + let last = b["messages"].as_array().and_then(|m| m.last()).and_then(|m| m["content"].as_str()).unwrap_or("").to_string(); + Json(json!({"choices": [{"message": {"role": "assistant", "content": format!("echo[{}]: {last}", b["model"].as_str().unwrap_or(""))}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 3, "completion_tokens": 2}})) + }), + ); + let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = l.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(l, app).await.unwrap() }); + format!("http://{addr}") + } + + #[tokio::test] + async fn answers_from_an_openai_compatible_runtime() { + let base = fake_runtime().await; + let models = list_models(&base).await.unwrap(); + assert_eq!(models, vec!["tiny:1b".to_string()], "embedding models are not chat models"); + let s = json!({"v": 1, "messages": [{"role": "user", "content": "ping"}]}); + let r = execute(&base, &s, &models).await; + assert!(valid_result(&r), "{r}"); + assert_eq!(r["text"], "echo[tiny:1b]: ping"); + assert_eq!(r["model"], "tiny:1b"); + let wrong = json!({"v": 1, "model": "big:70b", "messages": [{"role": "user", "content": "ping"}]}); + let r = execute(&base, &wrong, &models).await; + assert_eq!(r["ok"], false); + assert!(r["error"].as_str().unwrap().contains("not served")); + // Runtime down: an ok:false answer, not a crash. + let r = execute("http://127.0.0.1:1", &s, &models).await; + assert_eq!(r["ok"], false); + assert!(list_models("http://127.0.0.1:1").await.is_err()); + } +} diff --git a/node/crates/c0mpute-core/src/jobs_gateway.rs b/node/crates/c0mpute-core/src/jobs_gateway.rs index 38afa6f7..6267031f 100644 --- a/node/crates/c0mpute-core/src/jobs_gateway.rs +++ b/node/crates/c0mpute-core/src/jobs_gateway.rs @@ -5,8 +5,9 @@ //! winner's receipt, fetches the result by the receipt's hash from the //! winner, checks the bytes against that hash, and answers with the result. //! Accepts `workflow.step`, `crawl.fetch`, `ocr.extract` (input by hash), -//! `embed.text` and `blender.render-frame` (input pieces by hash; the bridge -//! fetches and checks the frames and returns them base64-encoded); +//! `embed.text`, `blender.render-frame` (input pieces by hash; the bridge +//! fetches and checks the frames and returns them base64-encoded) and +//! `infernet.inference` (chat completions); //! [`Workload`] is where more go. //! //! Same hardening as `c0mpute speech serve`: a required bearer token @@ -55,6 +56,7 @@ enum Workload { OcrExtract, EmbedText, RenderFrame, + Inference, } impl Workload { @@ -64,6 +66,7 @@ impl Workload { Workload::OcrExtract, Workload::EmbedText, Workload::RenderFrame, + Workload::Inference, ]; fn parse(name: &str) -> Option { @@ -76,6 +79,7 @@ impl Workload { Workload::OcrExtract => crate::ocr::WORKLOAD, Workload::EmbedText => crate::embed::WORKLOAD, Workload::RenderFrame => crate::render::WORKLOAD, + Workload::Inference => crate::inference::WORKLOAD, } } /// Reject a spec no worker could run before auctioning it. `workflow.step` @@ -87,6 +91,7 @@ impl Workload { Workload::OcrExtract => crate::ocr::parse_spec(spec).map(|_| ()), Workload::EmbedText => crate::embed::parse_spec(spec).map(|_| ()), Workload::RenderFrame => crate::render::parse_spec(spec).map(|_| ()), + Workload::Inference => crate::inference::parse_spec(spec).map(|_| ()), } } fn required_capabilities(self, spec: &Value) -> Vec { @@ -104,6 +109,7 @@ impl Workload { ] } Workload::RenderFrame => crate::render::required_capabilities_for(spec), + Workload::Inference => crate::inference::required_capabilities(spec), } } fn max_result_bytes(self) -> usize { @@ -112,6 +118,7 @@ impl Workload { Workload::OcrExtract => crate::ocr::MAX_RESULT_BYTES, Workload::EmbedText => crate::embed::MAX_RESULT_BYTES, Workload::RenderFrame => crate::render::MAX_RESULT_BYTES, + Workload::Inference => crate::inference::MAX_RESULT_BYTES, } } fn valid_result(self, spec: &Value, value: &Value) -> bool { @@ -126,6 +133,7 @@ impl Workload { }), Workload::RenderFrame => crate::render::parse_spec(spec) .is_ok_and(|parsed| crate::render::valid_result(&parsed, value)), + Workload::Inference => crate::inference::valid_result(value), } } } diff --git a/node/crates/c0mpute-core/src/lib.rs b/node/crates/c0mpute-core/src/lib.rs index 0bc26f3f..76dc2082 100644 --- a/node/crates/c0mpute-core/src/lib.rs +++ b/node/crates/c0mpute-core/src/lib.rs @@ -26,6 +26,7 @@ pub mod podcast; pub mod pricing; pub mod reachability; pub mod render; +pub mod inference; pub mod private; pub mod register; pub mod resources; diff --git a/node/crates/c0mpute-core/src/pieces.rs b/node/crates/c0mpute-core/src/pieces.rs index cdae9e1c..e7d47f74 100644 --- a/node/crates/c0mpute-core/src/pieces.rs +++ b/node/crates/c0mpute-core/src/pieces.rs @@ -334,6 +334,56 @@ impl InputRef { } } + /// The ref for stored [`Pieces`] (one piece: a single chunk). + pub fn of(p: &Pieces) -> Result { + let whole = Hash::from_hex(&p.hash).context("piece hash")?; + if p.pieces.len() <= 1 { + return Ok(Self::Single(whole)); + } + Ok(Self::Pieces { + whole, + size: p.size, + pieces: p + .pieces + .iter() + .map(|h| Hash::from_hex(h)) + .collect::, _>>() + .context("piece hash")?, + }) + } + + /// Fetch and verify the whole input into `dest` (streamed for pieces: + /// never whole in memory), renamed into place only once every check passed. + pub async fn fetch_to_file( + &self, + net: &Arc, + provider: Option, + max_bytes: usize, + dest: &Path, + ) -> Result<()> { + match self { + Self::Single(_) => { + let bytes = self.fetch(net, provider, max_bytes).await?; + let tmp = dest.with_extension(format!("part-{}", uuid::Uuid::new_v4())); + tokio::fs::write(&tmp, &bytes).await?; + tokio::fs::rename(&tmp, dest).await?; + Ok(()) + } + Self::Pieces { + whole, + size, + pieces, + } => { + if *size as usize > max_bytes { + bail!("the input is over the {max_bytes}-byte limit"); + } + fetch_pieces_to_file(net, provider, pieces, *size, *whole, INPUT_PIECE_BYTES, dest) + .await + .context("fetch the input pieces") + } + } + } + /// Fetch and verify the whole input into memory. pub async fn fetch( &self, diff --git a/node/crates/c0mpute-core/src/status_aggregator.rs b/node/crates/c0mpute-core/src/status_aggregator.rs index edf1cfe2..0bac1d0e 100644 --- a/node/crates/c0mpute-core/src/status_aggregator.rs +++ b/node/crates/c0mpute-core/src/status_aggregator.rs @@ -55,8 +55,8 @@ const DHT_CRAWL_INTERVAL: Duration = Duration::from_secs(30); /// Workload job topics we subscribe to. Gossipsub can't wildcard, so we /// enumerate the known workload types. `ffmpeg.transcode` is emitted by the -/// worker daemon; `infernet.inference` is emitted by the infernet peer -/// binary. Both always appear in the output (as zeros when idle) to mirror +/// worker daemon; `infernet.inference` by workers serving their +/// local OpenAI-compatible runtime. Both always appear in the output (as zeros when idle) to mirror /// the page's stub shape. // The temporal workloads (DIP 0026) are listed so seeds running the // aggregator also relay them for buyers that only know the seeds; their diff --git a/node/crates/c0mpute-core/src/supervisor.rs b/node/crates/c0mpute-core/src/supervisor.rs index 80e4f4a5..b56ebcf4 100644 --- a/node/crates/c0mpute-core/src/supervisor.rs +++ b/node/crates/c0mpute-core/src/supervisor.rs @@ -321,6 +321,29 @@ impl Supervisor { }); continue; } + // Chat completions from the operator's own OpenAI-compatible + // runtime. Probe it now and every minute; bid only while it lists + // chat models (the offer filter re-checks), advertise likewise. + if workload_type == crate::inference::WORKLOAD { + let Some(base) = crate::inference::runtime_url() else { + info!(%workload_type, "not subscribing: {} is off", crate::inference::RUNTIME_URL_ENV); + continue; + }; + tokio::spawn(crate::inference::refresh_loop(base.clone())); + let load = crate::inference::worker_load(); + let runner_tx = crate::inference::spawn_inference_runner(net.clone(), self.store.clone(), base, load.clone()); + info!(%workload_type, "subscribing to job topic"); + dispatch::run_worker_subscriber_with_load( + net.clone(), + workload_type, + tags.clone(), + Some(runner_tx), + Some(crate::inference::offer_filter()), + Some(self.config.pricing.clone()), + Some(load), + ); + continue; + } // Embeddings: provision fastembed in the background (never blocks // the worker), subscribe once a model answers. Same shape as TTS. if workload_type == crate::embed::WORKLOAD { @@ -655,7 +678,8 @@ pub fn which_ffmpeg() -> Option { None } -struct StoreSource(ChunkStore); +/// Serves chunk fetches from a [`ChunkStore`]. +pub struct StoreSource(pub ChunkStore); #[async_trait::async_trait] impl ChunkSource for StoreSource { diff --git a/node/crates/c0mpute-core/src/temporal/node.rs b/node/crates/c0mpute-core/src/temporal/node.rs index 4ec78e54..0cfdf37a 100644 --- a/node/crates/c0mpute-core/src/temporal/node.rs +++ b/node/crates/c0mpute-core/src/temporal/node.rs @@ -8,6 +8,7 @@ use std::time::Duration; use anyhow::{Context, Result, bail}; use c0mpute_net::{Libp2pNetwork, Multiaddr, NetworkConfig, bootstrap}; +use c0mpute_store::ChunkStore; use tokio::sync::Semaphore; use tracing::info; @@ -27,6 +28,16 @@ pub struct NetOptions { } pub async fn spawn_node(o: &NetOptions) -> Result> { + spawn(o, None).await +} + +/// [`spawn_node`] that also answers chunk fetches from `store`, so peers can +/// pull pieces this node stored and announced (text-train datasets and adapters). +pub async fn spawn_node_with_store(o: &NetOptions, store: ChunkStore) -> Result> { + spawn(o, Some(store)).await +} + +async fn spawn(o: &NetOptions, store: Option) -> Result> { std::fs::create_dir_all(&o.identity_dir)?; let parse = |v: &[String]| -> Result> { v.iter().map(|s| s.parse::().with_context(|| format!("bad multiaddr {s}"))).collect() }; let mut boot = parse(&o.bootstrap)?; @@ -34,7 +45,10 @@ pub async fn spawn_node(o: &NetOptions) -> Result> { boot.extend(bootstrap::fetch_or_fallback(bootstrap::DEFAULT_BOOTSTRAP_URL, bootstrap::default_seeds()).await); } let listen = if o.listen.is_empty() { vec!["/ip4/0.0.0.0/tcp/0".to_string()] } else { o.listen.clone() }; - let cfg = NetworkConfig::for_dir(o.identity_dir.clone()).with_listen(parse(&listen)?).with_bootstrap(boot); + let mut cfg = NetworkConfig::for_dir(o.identity_dir.clone()).with_listen(parse(&listen)?).with_bootstrap(boot); + if let Some(store) = store { + cfg = cfg.with_local_source(Arc::new(crate::supervisor::StoreSource(store))); + } Ok(Arc::new(Libp2pNetwork::spawn(cfg).await?)) } diff --git a/node/crates/c0mpute-core/src/text.rs b/node/crates/c0mpute-core/src/text.rs index e673661d..e0cd458e 100644 --- a/node/crates/c0mpute-core/src/text.rs +++ b/node/crates/c0mpute-core/src/text.rs @@ -2,18 +2,26 @@ //! `infernet.text.infer` (answers from a trained adapter, further down). //! //! A buyer has a job file (`@infernetprotocol/training` job config, `kind: -//! "lora"`) and a JSONL corpus. `c0mpute infernet text train` serves the -//! corpus from a token-gated transfer server, auctions the job to workers that -//! advertise `infernet:text:train`, and shows the winner's training live: the -//! worker relays every training-protocol v1 event (`meta`, `step`, `eval`, -//! `checkpoint`, `done`) to the transfer server as it happens. The worker then -//! uploads the LoRA adapter, and the buyer keeps it only if its sha256 matches -//! the receipt's `output_hash`. +//! "lora"`) and a JSONL corpus. `c0mpute infernet text train` (and its front +//! door `c0mpute train lora`) stores the corpus as content-addressed pieces in +//! its own chunk store, auctions the job to workers that advertise +//! `infernet:text:train`, and shows the winner's training live: the worker +//! pulls the pieces over libp2p and publishes every training-protocol v1 +//! event (`meta`, `step`, `eval`, `checkpoint`, `done`) on the job topic as it +//! happens. The worker then stores the LoRA adapter as pieces behind an +//! [`AdapterManifest`] whose hash its receipt carries; the buyer fetches it +//! over libp2p and keeps it only if the weights' sha256 matches. Peer to peer +//! end to end: neither side needs a reachable HTTP address. +//! +//! Workers before the pieces transport take the corpus and return the adapter +//! over HTTP from the buyer's transfer server; buyers opt into serving them +//! with `--legacy-http` (see [`TextTrainSpec`]). //! //! The training itself is Infernet's (`infernet model train --job -//! --json`, the `peft` backend). c0mpute only moves the data, the progress and -//! the result, over the same offer → bid → accept → receipt protocol as every -//! other workload. One job per node at a time. +//! --json`, the `peft` backend); c0mpute hands it local file paths. c0mpute +//! only moves the data, the progress and the result, over the same offer → +//! bid → accept → receipt protocol as every other workload. One job per node +//! at a time. use std::collections::HashMap; use std::net::SocketAddr; @@ -23,7 +31,8 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use anyhow::{Context, Result, bail}; -use c0mpute_net::Libp2pNetwork; +use c0mpute_net::{Libp2pNetwork, Network, PeerId}; +use c0mpute_store::ChunkStore; use c0mpute_net::topics::JobStatus; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; @@ -36,7 +45,7 @@ use crate::buyer::JobAuction; use crate::runner::{AcceptedJob, publish_receipt}; use crate::temporal::client::auction_with_mesh_retry; use crate::temporal::http::{SlotEvent, TransferServer}; -use crate::temporal::node::{NetOptions, spawn_node}; +use crate::temporal::node::{NetOptions, spawn_node_with_store}; pub const TRAIN_WORKLOAD: &str = "infernet.text.train"; pub const KINDS: [&str; 2] = ["lora", "peft"]; @@ -49,19 +58,91 @@ pub fn tag(workload: &str) -> String { workload.replace('.', ":") } +/// Capability tag: this worker's runner trains `{"messages": [...]}` records +/// as whole multi-turn conversations (Infernet's peft runner reports +/// `formats: ["text", "messages"]`). Buyers whose corpus has conversations +/// require it, so an older runner (which would drop them) never wins. +pub const MESSAGES_FORMAT_TAG: &str = "infernet:text:format:messages"; +/// Largest adapter file a buyer accepts as pieces. +pub const MAX_ADAPTER_FILE_BYTES: usize = 2 << 30; +/// How long a worker keeps a finished adapter's pieces for the buyer to fetch. +const ADAPTER_PIECES_TTL: Duration = Duration::from_secs(6 * 3600); +/// Gossip message type a worker publishes each training event as. +pub const EVENT_MSG: &str = "text_train_event"; +/// At most one `step` event per this interval goes on the wire. +const STEP_EVENT_INTERVAL: Duration = Duration::from_millis(250); + /// `JobOffer.spec_inline` for `infernet.text.train`. +/// +/// Two transports, both checked against the same `dataset_hash` and adapter +/// hash: +/// +/// - **pieces (peer to peer, this version on)**: the buyer stores the corpus +/// as [`crate::pieces::INPUT_PIECE_BYTES`] pieces and names them in +/// `corpus_hash` / `corpus_size` / `corpus_pieces`; the worker pulls them +/// over libp2p. With `adapter_pieces` the worker stores the adapter files as +/// pieces too and its receipt's `output_hash` is the content hash of an +/// [`AdapterManifest`] chunk. With `run_id` it publishes every training +/// event on the job topic. Nothing needs to be dialable over HTTP. +/// - **HTTP (workers before this version)**: `dataset_url`, +/// `artifact_upload_url` and `events_url` on the buyer's transfer server; +/// the receipt's `output_hash` is `sha256:` of the adapter weights. A buyer +/// fills these only with `--legacy-http`. Older workers cannot parse a spec +/// without `dataset_url`, so they never bid on a pieces-only job. +/// +/// A worker that understands both prefers pieces. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] pub struct TextTrainSpec { /// The job config. The worker rewrites `dataset.url` and /// `output.destination_url` to local paths; everything else is Infernet's. pub job: Value, - pub dataset_url: String, /// `sha256:` of the corpus; the worker refuses a mismatch. pub dataset_hash: String, + #[serde(default, skip_serializing_if = "String::is_empty")] + pub dataset_url: String, + #[serde(default, skip_serializing_if = "String::is_empty")] pub artifact_upload_url: String, - /// Where the worker posts each training event as it happens. + /// Where the worker posts each training event as it happens (HTTP). #[serde(default, skip_serializing_if = "Option::is_none")] pub events_url: Option, + /// Content hash of the corpus as stored pieces (see [`crate::pieces::InputRef`]). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub corpus_hash: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub corpus_size: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub corpus_pieces: Option>, + /// Return the adapter as pieces behind an [`AdapterManifest`]. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub adapter_pieces: bool, + /// Publish training events on the job topic tagged with this id. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub run_id: Option, +} + +/// What a worker's receipt points at when the adapter travels as pieces. +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub struct AdapterManifest { + pub v: u32, + /// `sha256:` of adapter_model.safetensors (the checkpoint's `model`). + pub adapter_sha256: String, + /// Each of [`ARTIFACTS`] the worker produced. + pub files: std::collections::BTreeMap, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub struct ManifestFile { + pub hash: String, + pub size: u64, + pub pieces: Vec, +} + +/// A training event as published on the job topic. +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] +pub struct EventMsg { + pub t: String, + pub run_id: String, + pub event: Value, } fn is_http(u: &str) -> bool { @@ -82,11 +163,48 @@ impl TextTrainSpec { if !crate::temporal::is_sha256_ref(&self.dataset_hash) { bail!("dataset_hash must be sha256:<64 lowercase hex>"); } - if !is_http(&self.dataset_url) || !is_http(&self.artifact_upload_url) || self.events_url.as_deref().is_some_and(|u| !is_http(u)) { + let urls = [Some(self.dataset_url.as_str()), Some(self.artifact_upload_url.as_str()), self.events_url.as_deref()]; + if urls.into_iter().flatten().any(|u| !u.is_empty() && !is_http(u)) { bail!("dataset, upload and events URLs must be http(s)"); } + if self.corpus_ref().transpose()?.is_none() && self.dataset_url.is_empty() { + bail!("the spec names no corpus: corpus pieces or a dataset URL"); + } + if !self.adapter_pieces && self.artifact_upload_url.is_empty() { + bail!("the spec names no way back for the adapter: adapter_pieces or an upload URL"); + } + if self.run_id.as_ref().is_some_and(|r| r.is_empty() || r.len() > 64 || !r.chars().all(|c| c.is_ascii_alphanumeric() || c == '-')) { + bail!("run_id must be 1-64 letters, digits or '-'"); + } Ok(()) } + + /// The corpus as pieces, when the spec names them. + pub fn corpus_ref(&self) -> Option> { + let hash = self.corpus_hash.as_deref()?; + Some( + crate::pieces::InputRef::parse("corpus", hash, self.corpus_size, self.corpus_pieces.as_deref(), MAX_DATASET as usize) + .map_err(anyhow::Error::msg), + ) + } +} + +/// Capability tags a worker needs to train `corpus`. +pub fn required_tags(corpus: &[u8]) -> Vec { + let mut t = vec![tag(TRAIN_WORKLOAD)]; + if corpus_has_conversations(corpus) { + t.push(MESSAGES_FORMAT_TAG.into()); + } + t +} + +/// Does any record of a JSONL corpus carry `messages` (a conversation)? +pub fn corpus_has_conversations(corpus: &[u8]) -> bool { + corpus.split(|b| *b == b'\n').any(|line| { + // Cheap pre-check before parsing: most corpora are {title, text}. + line.windows(10).any(|w| w == b"\"messages\"") + && serde_json::from_slice::(line).is_ok_and(|v| v.get("messages").is_some_and(Value::is_array)) + }) } pub fn sha256_hex(b: &[u8]) -> String { @@ -126,6 +244,11 @@ pub fn worker_tags(runtime: &Value, ram_mb: u64) -> Vec { if runtime["cuda"] == json!(true) { t.push("infernet:text:device:cuda".into()); } + // Infernet's runner trains whole conversations from the release that + // reports it; older ones would silently keep only {title, text} records. + if runtime["formats"].as_array().is_some_and(|f| f.iter().any(|x| x == "messages")) { + t.push(MESSAGES_FORMAT_TAG.into()); + } for gb in crate::temporal::caps::RAM_BUCKETS_GB { if ram_mb >= gb * 1024 { t.push(format!("c0mpute:ram:{gb}gb+")); @@ -151,13 +274,32 @@ async fn download(url: &str, dst: &Path, want: &str) -> Result<()> { Ok(()) } +/// Fetch the corpus into `dst`: pieces over libp2p (from the buyer first) +/// when the spec names them, else the legacy HTTP URL. Either way the file's +/// sha256 must be the offer's `dataset_hash`. +pub async fn fetch_corpus(net: &Arc, buyer: Option, spec: &TextTrainSpec, dst: &Path) -> Result<()> { + match spec.corpus_ref() { + Some(r) => { + r?.fetch_to_file(net, buyer, MAX_DATASET as usize, dst).await.context("fetch corpus pieces")?; + let got = crate::temporal::worker::file_sha256(dst)?; + if got != spec.dataset_hash { + let _ = tokio::fs::remove_file(dst).await; + bail!("dataset hash {got} differs from the offer's {}", spec.dataset_hash); + } + Ok(()) + } + None => download(&spec.dataset_url, dst, &spec.dataset_hash).await, + } +} + /// Run one accepted job: fetch the corpus, run `infernet model train --job /// --json`, relay every event, return the adapter dir and its weights hash. -pub async fn run_train(infernet: &Path, job_dir: &Path, spec: &TextTrainSpec) -> Result<(PathBuf, String)> { +/// `buyer` is the peer the corpus pieces are fetched from first. +pub async fn run_train(infernet: &Path, job_dir: &Path, spec: &TextTrainSpec, net: &Arc, buyer: Option) -> Result<(PathBuf, String)> { spec.validate()?; tokio::fs::create_dir_all(job_dir).await?; let dataset = job_dir.join("dataset.jsonl"); - download(&spec.dataset_url, &dataset, &spec.dataset_hash).await.context("fetch dataset")?; + fetch_corpus(net, buyer, spec, &dataset).await.context("fetch dataset")?; let mut job = spec.job.clone(); job["dataset"] = json!({"url": "dataset.jsonl", "sha256": spec.dataset_hash.trim_start_matches("sha256:")}); job["output"] = json!({"destination_url": "adapter"}); @@ -188,13 +330,34 @@ pub async fn run_train(infernet: &Path, job_dir: &Path, spec: &TextTrainSpec) -> last }); - // Events go to the buyer from their own task so a slow link never stalls training. + // Events go to the buyer from their own task so a slow link never stalls + // training: on the job topic (p2p, `run_id`) and/or to the legacy HTTP URL. let (ev_tx, mut ev_rx) = mpsc::unbounded_channel::(); - let relay = spec.events_url.clone().map(|url| { + let relaying = spec.events_url.is_some() || spec.run_id.is_some(); + let relay = relaying.then(|| { + let (url, run_id, net) = (spec.events_url.clone(), spec.run_id.clone(), net.clone()); tokio::spawn(async move { let c = reqwest::Client::builder().timeout(Duration::from_secs(10)).build().unwrap_or_default(); + let topic = c0mpute_net::topics::job_topic(TRAIN_WORKLOAD); + let mut last_step: Option = None; while let Some(ev) = ev_rx.recv().await { - if let Err(e) = c.post(&url).json(&ev).send().await { + if let Some(run_id) = &run_id { + let step = ev["type"] == "step"; + if !(step && last_step.is_some_and(|t| t.elapsed() < STEP_EVENT_INTERVAL)) { + if step { + last_step = Some(std::time::Instant::now()); + } + let msg = EventMsg { t: EVENT_MSG.into(), run_id: run_id.clone(), event: ev.clone() }; + if let Ok(b) = serde_json::to_vec(&msg) + && let Err(e) = net.publish(&topic, b).await + { + warn!(err = %e, "text: could not publish a training event"); + } + } + } + if let Some(url) = &url + && let Err(e) = c.post(url).json(&ev).send().await + { warn!(err = %e, "text: could not relay a training event"); } } @@ -215,7 +378,7 @@ pub async fn run_train(infernet: &Path, job_dir: &Path, spec: &TextTrainSpec) -> Some("done") => done = Some(ev.clone()), _ => {} } - if spec.events_url.is_some() { + if relaying { let _ = ev_tx.send(ev); } } @@ -253,21 +416,61 @@ async fn upload(base: &str, dir: &Path) -> Result<()> { Ok(()) } -pub fn spawn_train_runner(net: Arc, cache_root: PathBuf, infernet: PathBuf, busy: Arc, slots: Arc) -> mpsc::Sender { +/// Store the adapter files as pieces, then the [`AdapterManifest`] naming +/// them; returns the manifest's content hash (what the receipt carries). The +/// pieces are dropped from the store after [`ADAPTER_PIECES_TTL`]. +pub async fn store_adapter(net: &Arc, store: &ChunkStore, dir: &Path, adapter_sha256: &str) -> Result { + let mut files = std::collections::BTreeMap::new(); + let mut held = Vec::new(); + for f in ARTIFACTS { + let p = dir.join(f); + if !p.exists() { + continue; + } + let stored = crate::pieces::put_file_pieces(net, store, &p, MAX_ADAPTER_FILE_BYTES as u64, crate::pieces::INPUT_PIECE_BYTES).await.with_context(|| format!("store {f}"))?; + held.extend(stored.pieces.iter().cloned()); + files.insert(f.to_string(), ManifestFile { hash: stored.hash, size: stored.size, pieces: stored.pieces }); + } + if !files.contains_key("adapter_model.safetensors") { + bail!("no adapter_model.safetensors to store"); + } + let manifest = serde_json::to_vec(&AdapterManifest { v: 1, adapter_sha256: adapter_sha256.to_string(), files })?; + let hash = store.put(&manifest).await.context("store adapter manifest")?; + net.announce(&hash).await.context("announce adapter manifest")?; + held.push(hash.to_hex()); + let store = store.clone(); + tokio::spawn(async move { + tokio::time::sleep(ADAPTER_PIECES_TTL).await; + for h in held { + if let Ok(h) = c0mpute_proto::Hash::from_hex(&h) { + let _ = store.delete(&h).await; + } + } + }); + Ok(hash.to_hex()) +} + +pub fn spawn_train_runner(net: Arc, store: ChunkStore, cache_root: PathBuf, infernet: PathBuf, busy: Arc, slots: Arc) -> mpsc::Sender { let (tx, mut rx) = mpsc::channel::(8); tokio::spawn(async move { while let Some(job) = rx.recv().await { let Ok(_slot) = slots.clone().acquire_owned().await else { return }; let accept = job.accept.clone(); busy.store(true, Ordering::Relaxed); + let job_dir = cache_root.join("text").join("jobs").join(&accept.job_id); let result: Result = async { let spec: TextTrainSpec = serde_json::from_value(accept.spec_inline.clone().context("accept lacks spec_inline")?)?; - let job_dir = cache_root.join("text").join("jobs").join(&accept.job_id); - let (dir, hash) = run_train(&infernet, &job_dir, &spec).await?; + let buyer = accept.buyer_peer_id.parse::().ok(); + let (dir, hash) = run_train(&infernet, &job_dir, &spec, &net, buyer).await?; + if spec.adapter_pieces { + // Receipt -> manifest -> pieces, all over libp2p. + return store_adapter(&net, &store, &dir, &hash).await; + } upload(&spec.artifact_upload_url, &dir).await.context("upload adapter")?; Ok(hash) } .await; + let _ = tokio::fs::remove_dir_all(&job_dir).await; busy.store(false, Ordering::Relaxed); match result { Ok(hash) => { @@ -309,9 +512,17 @@ const MAX_PROMPT_BYTES: usize = 32 * 1024; const ANSWER_SLOT: &str = "answer"; /// `JobOffer.spec_inline` for `infernet.text.infer`: one answer from a base -/// model + LoRA adapter. The adapter travels with the request (a bundle on -/// the buyer's transfer server) and is cached by hash, so any worker can -/// serve any adapter; the buyer does retrieval, the worker only generates. +/// model + LoRA adapter. The adapter travels with the request and is cached +/// by hash, so any worker can serve any adapter; the buyer does retrieval, +/// the worker only generates. +/// +/// Peer to peer (this version on): the buyer stores the adapter as pieces +/// behind an [`AdapterManifest`] (`adapter_manifest`), the worker fetches it +/// over libp2p on a cache miss, and with `answer_chunk` stores `answer.json` +/// as a chunk whose hash its receipt carries. Workers before this version need +/// `adapter_url` / `answer_upload_url` on the buyer's transfer server (the +/// buyer's `--legacy-http`); they cannot parse a spec without them, so they +/// never bid on a pieces-only request. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] pub struct TextInferSpec { /// `sha256:` of adapter_model.safetensors. @@ -320,9 +531,17 @@ pub struct TextInferSpec { pub messages: Vec, pub max_new_tokens: u32, /// `{"files": {"": ""}}` for the adapter, fetched only on a cache miss. + #[serde(default, skip_serializing_if = "String::is_empty")] pub adapter_url: String, /// Base URL the worker PUTs `answer.json` to. + #[serde(default, skip_serializing_if = "String::is_empty")] pub answer_upload_url: String, + /// Content hash of the adapter's [`AdapterManifest`] chunk. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub adapter_manifest: Option, + /// Return `answer.json` as a chunk (receipt `output_hash` = its content hash). + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub answer_chunk: bool, } fn valid_repo_id(base: &str) -> bool { @@ -354,9 +573,18 @@ impl TextInferSpec { if !(1..=1024).contains(&self.max_new_tokens) { bail!("max_new_tokens must be 1..=1024"); } - if !is_http(&self.adapter_url) || !is_http(&self.answer_upload_url) { + if [&self.adapter_url, &self.answer_upload_url].iter().any(|u| !u.is_empty() && !is_http(u)) { bail!("adapter and answer URLs must be http(s)"); } + if self.adapter_manifest.as_deref().is_some_and(|h| c0mpute_proto::Hash::from_hex(h).is_err()) { + bail!("adapter_manifest must be a hex content hash"); + } + if self.adapter_manifest.is_none() && self.adapter_url.is_empty() { + bail!("the spec names no adapter: adapter_manifest or adapter_url"); + } + if !self.answer_chunk && self.answer_upload_url.is_empty() { + bail!("the spec names no way back for the answer: answer_chunk or an upload URL"); + } Ok(()) } } @@ -407,12 +635,23 @@ async fn fetch_adapter(url: &str, want: &str, dir: &Path) -> Result<()> { } /// Answer one request: adapter from the cache (or fetched), then `infernet model generate`. -pub async fn run_infer(infernet: &Path, cache_root: &Path, job_id: &str, spec: &TextInferSpec) -> Result> { +/// `buyer` is the peer adapter pieces are fetched from first. +pub async fn run_infer(infernet: &Path, cache_root: &Path, job_id: &str, spec: &TextInferSpec, net: &Arc, buyer: Option) -> Result> { spec.validate()?; let adapter = cache_root.join("text").join("adapters").join(spec.adapter_hash.trim_start_matches("sha256:")); let cached = crate::temporal::worker::file_sha256(&adapter.join("adapter_model.safetensors")).ok(); if cached.as_deref() != Some(spec.adapter_hash.as_str()) { - fetch_adapter(&spec.adapter_url, &spec.adapter_hash, &adapter).await.context("fetch adapter")?; + match &spec.adapter_manifest { + Some(m) => { + let _ = tokio::fs::remove_dir_all(&adapter).await; + let got = fetch_adapter_pieces(net, buyer, m, &adapter).await.context("fetch adapter pieces")?; + if got != spec.adapter_hash { + let _ = tokio::fs::remove_dir_all(&adapter).await; + bail!("adapter {got} differs from the offer's {}", spec.adapter_hash); + } + } + None => fetch_adapter(&spec.adapter_url, &spec.adapter_hash, &adapter).await.context("fetch adapter")?, + } } let dir = cache_root.join("text").join("infer").join(job_id); tokio::fs::create_dir_all(&dir).await?; @@ -434,7 +673,7 @@ pub async fn run_infer(infernet: &Path, cache_root: &Path, job_id: &str, spec: & Ok(serde_json::to_vec(&ans)?) } -pub fn spawn_infer_runner(net: Arc, cache_root: PathBuf, infernet: PathBuf, busy: Arc) -> mpsc::Sender { +pub fn spawn_infer_runner(net: Arc, store: ChunkStore, cache_root: PathBuf, infernet: PathBuf, busy: Arc) -> mpsc::Sender { let (tx, mut rx) = mpsc::channel::(16); tokio::spawn(async move { while let Some(job) = rx.recv().await { @@ -442,7 +681,13 @@ pub fn spawn_infer_runner(net: Arc, cache_root: PathBuf, infernet busy.store(true, Ordering::Relaxed); let result: Result = async { let spec: TextInferSpec = serde_json::from_value(accept.spec_inline.clone().context("accept lacks spec_inline")?)?; - let answer = run_infer(&infernet, &cache_root, &accept.job_id, &spec).await?; + let buyer = accept.buyer_peer_id.parse::().ok(); + let answer = run_infer(&infernet, &cache_root, &accept.job_id, &spec, &net, buyer).await?; + if spec.answer_chunk { + let hash = store.put(&answer).await.context("store answer")?; + net.announce(&hash).await.context("announce answer")?; + return Ok(hash.to_hex()); + } reqwest::Client::new().put(format!("{}/answer.json", spec.answer_upload_url)).body(answer.clone()).send().await?.error_for_status()?; Ok(sha256_hex(&answer)) } @@ -500,16 +745,18 @@ pub async fn serve(net_opts: &NetOptions, infernet: PathBuf, cache_root: PathBuf if infer { tags.push(tag(INFER_WORKLOAD)); } - let net = spawn_node(net_opts).await?; + // Adapter pieces live here until the buyer has fetched them. + let store = ChunkStore::open(cache_root.join("text").join("store")).await?; + let net = spawn_node_with_store(net_opts, store.clone()).await?; let hardware = json!({"version": env!("CARGO_PKG_VERSION"), "text_runtime": "peft", "ram_mb": ram_mb, "cuda": runtime["cuda"]}); tokio::spawn(crate::capabilities::advertise_loop(net.clone(), tags.clone(), hardware, crate::capabilities::local_bench_score(), crate::capabilities::DEFAULT_ADVERTISE_INTERVAL)); let busy = Arc::new(AtomicBool::new(false)); let slots = Arc::new(Semaphore::new(1)); - let tx = spawn_train_runner(net.clone(), cache_root.clone(), infernet.clone(), busy.clone(), slots); + let tx = spawn_train_runner(net.clone(), store.clone(), cache_root.clone(), infernet.clone(), busy.clone(), slots); crate::dispatch::run_worker_subscriber_filtered(net.clone(), TRAIN_WORKLOAD.into(), tags.clone(), Some(tx), Some(train_offer_filter(busy))); if infer { let ibusy = Arc::new(AtomicBool::new(false)); - let itx = spawn_infer_runner(net.clone(), cache_root, infernet, ibusy.clone()); + let itx = spawn_infer_runner(net.clone(), store.clone(), cache_root, infernet, ibusy.clone()); crate::dispatch::run_worker_subscriber_filtered(net.clone(), INFER_WORKLOAD.into(), tags.clone(), Some(itx), Some(infer_offer_filter(ibusy))); } info!(peer = %net.peer_id().to_base58(), ?tags, "text: serving"); @@ -529,6 +776,10 @@ pub struct TrainRequest { pub max_price_usd: f64, pub gpu: bool, pub min_ram_gb: Option, + /// Also serve the corpus and take the adapter over HTTP from this machine, + /// so workers older than the pieces transport can bid (they need to reach + /// `public_url` / `bind`). Off by default: the job is fully peer to peer. + pub legacy_http: bool, pub public_url: Option, pub bind: SocketAddr, pub timeout_s: u64, @@ -602,19 +853,43 @@ pub fn render(ev: &Value, total: &mut Option) -> Option { } /// Auction the job, stream its progress through `on_event`, verify and keep the adapter. -pub async fn train(net: Arc, req: &TrainRequest, mut on_event: impl FnMut(&Value) + Send) -> Result { +/// +/// Peer to peer: the corpus goes out as pieces from `store` (served by `net`), +/// progress arrives on the job topic, and the adapter comes back as pieces +/// named by the manifest the worker's receipt points at. With +/// `req.legacy_http` the buyer also runs the HTTP transfer server, so a worker +/// that predates pieces can win; its adapter then arrives over HTTP and is +/// checked against the receipt exactly as before. +pub async fn train(net: Arc, store: &ChunkStore, req: &TrainRequest, mut on_event: impl FnMut(&Value) + Send) -> Result { let (job, bytes, preview) = load(req)?; - let (ev_tx, mut ev_rx) = mpsc::unbounded_channel::(); - let server = TransferServer::start_with(req.bind, req.public_url.clone(), bytes, &ARTIFACTS, Some(ev_tx)).await?; - let spec = TextTrainSpec { + let stored = crate::pieces::put_bytes_pieces(&net, store, &bytes, crate::pieces::INPUT_PIECE_BYTES).await.context("store the corpus as pieces")?; + let run_id = uuid::Uuid::new_v4().to_string(); + let mut spec = TextTrainSpec { job, - dataset_url: server.dataset_url(), dataset_hash: preview.dataset_hash.clone(), - artifact_upload_url: server.upload_url(SLOT), - events_url: Some(server.events_url(SLOT)), + dataset_url: String::new(), + artifact_upload_url: String::new(), + events_url: None, + corpus_hash: Some(stored.hash.clone()), + corpus_size: (stored.pieces.len() > 1).then_some(stored.size), + corpus_pieces: (stored.pieces.len() > 1).then(|| stored.pieces.clone()), + adapter_pieces: true, + run_id: Some(run_id.clone()), + }; + let (ev_tx, mut ev_rx) = mpsc::unbounded_channel::(); + let server = if req.legacy_http { + let s = TransferServer::start_with(req.bind, req.public_url.clone(), bytes.clone(), &ARTIFACTS, Some(ev_tx)).await?; + spec.dataset_url = s.dataset_url(); + spec.artifact_upload_url = s.upload_url(SLOT); + spec.events_url = Some(s.events_url(SLOT)); + Some(s) + } else { + drop(ev_tx); + None }; spec.validate()?; - let mut tags = vec![tag(TRAIN_WORKLOAD)]; + let mut tags = required_tags(&bytes); + drop(bytes); if req.gpu { tags.push("infernet:text:device:cuda".into()); } @@ -622,6 +897,16 @@ pub async fn train(net: Arc, req: &TrainRequest, mut on_event: im let b = crate::temporal::caps::RAM_BUCKETS_GB.iter().copied().find(|b| *b >= min).unwrap_or(128); tags.push(format!("c0mpute:ram:{b}gb+")); } + // Training events published on the job topic for this run. + let topic = c0mpute_net::topics::job_topic(TRAIN_WORKLOAD); + let mut gossip = net.messages(); + let p2p_event = |m: &c0mpute_net::GossipMessage| -> Option { + if m.topic != topic { + return None; + } + let msg: EventMsg = serde_json::from_slice(&m.data).ok()?; + (msg.t == EVENT_MSG && msg.run_id == run_id).then_some(msg.event) + }; let spec_v = serde_json::to_value(&spec)?; let mut last_err = None; let mut outcome = None; @@ -642,11 +927,21 @@ pub async fn train(net: Arc, req: &TrainRequest, mut on_event: im tokio::select! { r = &mut auction => break r, Some((_, ev)) = ev_rx.recv() => on_event(&ev), + m = gossip.recv() => match m { + Ok(m) => if let Some(ev) = p2p_event(&m) { on_event(&ev) }, + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {} + Err(_) => {} + }, } }; while let Ok((_, ev)) = ev_rx.try_recv() { on_event(&ev); } + while let Ok(m) = gossip.try_recv() { + if let Some(ev) = p2p_event(&m) { + on_event(&ev); + } + } match res { Ok(o) if o.receipt.status == JobStatus::Completed => { outcome = Some(o); @@ -656,22 +951,84 @@ pub async fn train(net: Arc, req: &TrainRequest, mut on_event: im Err(e) => last_err = Some(e.context(format!("attempt {}", attempt + 1))), } } - let o = outcome.ok_or_else(|| last_err.unwrap_or_else(|| anyhow::anyhow!("training failed")))?; - let files: HashMap> = server.take(SLOT).await; - let weights = files.get("adapter_model.safetensors").context("worker did not upload adapter_model.safetensors")?; - let got = sha256_hex(weights); - let claimed = o.receipt.output_hash.clone().unwrap_or_default(); - if got != claimed { - bail!("adapter {got} does not match the receipt's {claimed}"); + let result = async { + let o = outcome.ok_or_else(|| last_err.unwrap_or_else(|| anyhow::anyhow!("training failed")))?; + let claimed = o.receipt.output_hash.clone().unwrap_or_default(); + let (got, transport) = if claimed.starts_with("sha256:") { + // A worker before pieces: it uploaded over HTTP. + let server = server.as_ref().context("the worker answered with an HTTP upload, but this job has no HTTP transfer server")?; + let files: HashMap> = server.take(SLOT).await; + let weights = files.get("adapter_model.safetensors").context("worker did not upload adapter_model.safetensors")?; + let got = sha256_hex(weights); + if got != claimed { + bail!("adapter {got} does not match the receipt's {claimed}"); + } + std::fs::create_dir_all(&req.output)?; + for (name, data) in &files { + std::fs::write(req.output.join(name), data)?; + } + (got, "http") + } else { + let worker = o.receipt.worker_peer_id.parse::().ok(); + (fetch_adapter_pieces(&net, worker, &claimed, &req.output).await?, "pieces") + }; + Ok(json!({ + "adapter_dir": req.output, "adapter_hash": got, "base_model": preview.base_model, "dataset_hash": preview.dataset_hash, + "transport": transport, + "receipt": {"job_id": o.receipt.job_id, "worker_peer_id": o.receipt.worker_peer_id, "output_hash": o.receipt.output_hash, "price_usd": o.accepted_bid.price_usd} + })) } - std::fs::create_dir_all(&req.output)?; - for (name, data) in &files { - std::fs::write(req.output.join(name), data)?; + .await; + // The corpus pieces have done their job (stored whole too when small). + for h in stored.pieces.iter().chain(std::iter::once(&stored.hash)) { + if let Ok(h) = c0mpute_proto::Hash::from_hex(h) { + let _ = store.delete(&h).await; + } + } + result +} + +/// Fetch the [`AdapterManifest`] a receipt names, then every file it lists, +/// into `out`; keep them only if adapter_model.safetensors hashes to the +/// manifest's `adapter_sha256`. Returns that hash. +pub async fn fetch_adapter_pieces(net: &Arc, worker: Option, manifest_hash: &str, out: &Path) -> Result { + let mref = crate::pieces::InputRef::parse("adapter_manifest", manifest_hash, None, None, 1 << 20).map_err(anyhow::Error::msg)?; + let manifest: AdapterManifest = serde_json::from_slice(&mref.fetch(net, worker, 1 << 20).await.context("fetch the adapter manifest")?).context("adapter manifest is not JSON")?; + if manifest.v != 1 || !crate::temporal::is_sha256_ref(&manifest.adapter_sha256) { + bail!("adapter manifest has an unknown version or a bad adapter_sha256"); + } + if !manifest.files.contains_key("adapter_model.safetensors") || manifest.files.keys().any(|k| !ARTIFACTS.contains(&k.as_str())) { + bail!("adapter manifest must list adapter_model.safetensors and only {ARTIFACTS:?}"); + } + let tmp = out.with_extension(format!("partial-{}", uuid::Uuid::new_v4())); + tokio::fs::create_dir_all(&tmp).await?; + let fetched: Result = async { + for (name, f) in &manifest.files { + let (size, pieces) = if f.pieces.len() > 1 { (Some(f.size), Some(f.pieces.as_slice())) } else { (None, None) }; + let r = crate::pieces::InputRef::parse("adapter", &f.hash, size, pieces, MAX_ADAPTER_FILE_BYTES).map_err(anyhow::Error::msg)?; + r.fetch_to_file(net, worker, MAX_ADAPTER_FILE_BYTES, &tmp.join(name)).await.with_context(|| format!("fetch {name}"))?; + } + let got = crate::temporal::worker::file_sha256(&tmp.join("adapter_model.safetensors"))?; + if got != manifest.adapter_sha256 { + bail!("adapter {got} does not match the manifest's {}", manifest.adapter_sha256); + } + Ok(got) + } + .await; + match fetched { + Ok(got) => { + tokio::fs::create_dir_all(out).await?; + for name in manifest.files.keys() { + tokio::fs::rename(tmp.join(name), out.join(name)).await?; + } + let _ = tokio::fs::remove_dir_all(&tmp).await; + Ok(got) + } + Err(e) => { + let _ = tokio::fs::remove_dir_all(&tmp).await; + Err(e) + } } - Ok(json!({ - "adapter_dir": req.output, "adapter_hash": got, "base_model": preview.base_model, "dataset_hash": preview.dataset_hash, - "receipt": {"job_id": o.receipt.job_id, "worker_peer_id": o.receipt.worker_peer_id, "output_hash": o.receipt.output_hash, "price_usd": o.accepted_bid.price_usd} - })) } // ── chat (buyer) ──────────────────────────────────────────────────────── @@ -723,6 +1080,8 @@ pub struct ChatRequest { pub max_new_tokens: u32, pub workers: Vec, pub max_price_usd: f64, + /// Also serve workers older than the pieces transport over HTTP. + pub legacy_http: bool, pub public_url: Option, pub bind: SocketAddr, pub timeout_s: u64, @@ -730,7 +1089,12 @@ pub struct ChatRequest { } /// Ask a worker; returns the answer, its sources and the receipt. -pub async fn chat(net: Arc, req: &ChatRequest) -> Result { +/// +/// Peer to peer: the adapter goes out as pieces from `store` (the worker +/// fetches them only on a cache miss) and the answer comes back as a chunk +/// named by the receipt. `req.legacy_http` adds the HTTP transfer server for +/// older workers. +pub async fn chat(net: Arc, store: &ChunkStore, req: &ChatRequest) -> Result { let (bundle, adapter_hash, manifest) = adapter_bundle(&req.adapter_dir)?; let base_model = manifest["base_model"].as_str().context("the adapter's manifest.json has no base_model")?.to_string(); let sources = match &req.search { @@ -742,14 +1106,24 @@ pub async fn chat(net: Arc, req: &ChatRequest) -> Result { messages.push(json!({"role": "system", "content": system})); } messages.push(json!({"role": "user", "content": chat_prompt(&req.question, &sources)})); - let server = TransferServer::start_with(req.bind, req.public_url.clone(), bundle, &ANSWER_ARTIFACTS, None).await?; - let spec = TextInferSpec { + let adapter_manifest = store_adapter(&net, store, &req.adapter_dir, &adapter_hash).await.context("store the adapter as pieces")?; + let mut spec = TextInferSpec { adapter_hash: adapter_hash.clone(), base_model: base_model.clone(), messages, max_new_tokens: req.max_new_tokens, - adapter_url: server.dataset_url(), - answer_upload_url: server.upload_url(ANSWER_SLOT), + adapter_url: String::new(), + answer_upload_url: String::new(), + adapter_manifest: Some(adapter_manifest), + answer_chunk: true, + }; + let server = if req.legacy_http { + let s = TransferServer::start_with(req.bind, req.public_url.clone(), bundle, &ANSWER_ARTIFACTS, None).await?; + spec.adapter_url = s.dataset_url(); + spec.answer_upload_url = s.upload_url(ANSWER_SLOT); + Some(s) + } else { + None }; spec.validate()?; let spec_v = serde_json::to_value(&spec)?; @@ -762,17 +1136,30 @@ pub async fn chat(net: Arc, req: &ChatRequest) -> Result { .with_deadline(Duration::from_secs(req.timeout_s + 120)) .with_receipt_timeout(Duration::from_secs(req.timeout_s + 60)) }; - let o = auction_with_mesh_retry(net, make).await?; + let o = auction_with_mesh_retry(net.clone(), make).await?; if o.receipt.status != JobStatus::Completed { bail!("worker {} reported {:?}", o.accepted_bid.bidder_peer_id, o.receipt.status); } - let files = server.take(ANSWER_SLOT).await; - let answer = files.get("answer.json").context("worker did not upload answer.json")?; - let got = sha256_hex(answer); - if Some(got.as_str()) != o.receipt.output_hash.as_deref() { - bail!("answer {got} does not match the receipt's {:?}", o.receipt.output_hash); + let claimed = o.receipt.output_hash.clone().unwrap_or_default(); + let answer = if claimed.starts_with("sha256:") { + // A worker before pieces: it uploaded over HTTP. + let server = server.as_ref().context("the worker answered over HTTP, but this request has no transfer server")?; + let files = server.take(ANSWER_SLOT).await; + let answer = files.get("answer.json").context("worker did not upload answer.json")?.clone(); + let got = sha256_hex(&answer); + if got != claimed { + bail!("answer {got} does not match the receipt's {claimed}"); + } + answer + } else { + // The fetch checks the bytes against the receipt's content hash. + let r = crate::pieces::InputRef::parse("answer", &claimed, None, None, 1 << 20).map_err(anyhow::Error::msg)?; + r.fetch(&net, o.receipt.worker_peer_id.parse::().ok(), 1 << 20).await.context("fetch the answer")? + }; + let ans: Value = serde_json::from_slice(&answer)?; + if ans["adapter"].as_str().is_some_and(|a| a != adapter_hash) { + bail!("the worker answered with adapter {} instead of {adapter_hash}", ans["adapter"]); } - let ans: Value = serde_json::from_slice(answer)?; Ok(json!({ "text": ans["text"], "sources": sources, "base_model": base_model, "adapter_hash": adapter_hash, "tokens": ans["tokens"], "seconds": ans["seconds"], @@ -792,7 +1179,33 @@ mod tests { max_new_tokens: 200, adapter_url: "http://127.0.0.1:1/t/x/dataset.json".into(), answer_upload_url: "http://127.0.0.1:1/t/x/a/answer".into(), + adapter_manifest: None, + answer_chunk: false, + } + } + + #[test] + fn p2p_infer_specs() { + let s = TextInferSpec { + adapter_url: String::new(), + answer_upload_url: String::new(), + adapter_manifest: Some(c0mpute_proto::Hash::of(b"m").to_hex()), + answer_chunk: true, + ..infer_spec() + }; + s.validate().unwrap(); + let v = serde_json::to_value(&s).unwrap(); + assert!(v.get("adapter_url").is_none()); + #[derive(Deserialize)] + #[allow(dead_code)] + struct OldInfer { + adapter_url: String, + answer_upload_url: String, } + assert!(serde_json::from_value::(v).is_err(), "old workers never bid on a pieces-only chat"); + assert!(TextInferSpec { adapter_manifest: None, ..s.clone() }.validate().is_err()); + assert!(TextInferSpec { answer_chunk: false, ..s.clone() }.validate().is_err()); + assert!(TextInferSpec { adapter_manifest: Some("zz".into()), ..s }.validate().is_err()); } #[test] @@ -847,9 +1260,74 @@ mod tests { dataset_hash: format!("sha256:{}", "a".repeat(64)), artifact_upload_url: "http://127.0.0.1:1/t/x/a/adapter".into(), events_url: Some("http://127.0.0.1:1/t/x/e/adapter".into()), + corpus_hash: None, + corpus_size: None, + corpus_pieces: None, + adapter_pieces: false, + run_id: None, } } + fn p2p_spec() -> TextTrainSpec { + TextTrainSpec { + dataset_url: String::new(), + artifact_upload_url: String::new(), + events_url: None, + corpus_hash: Some(c0mpute_proto::Hash::of(b"corpus").to_hex()), + adapter_pieces: true, + run_id: Some("0e1f2a3b-run".into()), + ..spec() + } + } + + #[test] + fn p2p_specs_validate_and_old_workers_cannot_parse_them() { + let s = p2p_spec(); + s.validate().unwrap(); + let v = serde_json::to_value(&s).unwrap(); + // No HTTP fields on the wire at all. + assert!(v.get("dataset_url").is_none() && v.get("artifact_upload_url").is_none() && v.get("events_url").is_none()); + // A worker before pieces requires dataset_url, so it never bids. + #[derive(Deserialize)] + #[allow(dead_code)] + struct OldSpec { + job: Value, + dataset_url: String, + dataset_hash: String, + artifact_upload_url: String, + } + assert!(serde_json::from_value::(v.clone()).is_err()); + assert_eq!(serde_json::from_value::(v).unwrap(), s); + // A legacy spec (an old buyer) still parses and validates. + let old = json!({"job": spec().job, "dataset_url": "http://h/d", "dataset_hash": spec().dataset_hash, "artifact_upload_url": "http://h/u"}); + serde_json::from_value::(old).unwrap().validate().unwrap(); + + let mut bad = p2p_spec(); + bad.corpus_hash = None; + assert!(bad.validate().unwrap_err().to_string().contains("no corpus")); + let mut bad = p2p_spec(); + bad.adapter_pieces = false; + assert!(bad.validate().unwrap_err().to_string().contains("way back")); + let mut bad = p2p_spec(); + bad.corpus_size = Some(3 << 20); + bad.corpus_pieces = Some(vec![c0mpute_proto::Hash::of(b"x").to_hex()]); + assert!(bad.validate().is_err(), "3 MiB cannot be one piece"); + let mut bad = p2p_spec(); + bad.run_id = Some("no spaces".into()); + assert!(bad.validate().is_err()); + } + + #[test] + fn conversations_require_the_messages_format_tag() { + let text = b"{\"title\":\"a\",\"text\":\"b\"}\n{\"title\":\"c\",\"text\":\"the word \\\"messages\\\" in text\"}\n"; + assert_eq!(required_tags(text), vec!["infernet:text:train".to_string()]); + let chat = b"{\"title\":\"a\",\"text\":\"b\"}\n{\"messages\":[{\"role\":\"user\",\"content\":\"hi\"},{\"role\":\"assistant\",\"content\":\"yo\"}]}\n"; + assert_eq!(required_tags(chat), vec!["infernet:text:train".to_string(), MESSAGES_FORMAT_TAG.to_string()]); + let t = worker_tags(&json!({"cuda": false, "formats": ["text", "messages"]}), 0); + assert!(t.contains(&MESSAGES_FORMAT_TAG.to_string())); + assert!(!worker_tags(&json!({"cuda": false}), 0).contains(&MESSAGES_FORMAT_TAG.to_string())); + } + #[test] fn validates_kind_model_hash_and_urls() { assert!(spec().validate().is_ok()); diff --git a/node/crates/c0mpute-core/src/train.rs b/node/crates/c0mpute-core/src/train.rs index c8c5b98a..1f725a40 100644 --- a/node/crates/c0mpute-core/src/train.rs +++ b/node/crates/c0mpute-core/src/train.rs @@ -2,22 +2,27 @@ //! training c0mpute already runs as `infernet.text.train` ([`crate::text`]). //! //! It does not train anything itself. It checks the buyer's dataset, turns -//! the common JSONL shapes into the `{title, text}` corpus the worker's peft -//! runner reads, writes the job config from a handful of flags, and hands -//! both to [`crate::text::train`], which auctions the job, relays progress, -//! and keeps the adapter only if its sha256 matches the receipt. +//! the common JSONL shapes into the corpus the worker's peft runner reads, +//! writes the job config from a handful of flags, and hands both to +//! [`crate::text::train`], which auctions the job peer to peer, relays +//! progress, and keeps the adapter only if its sha256 matches. //! //! Accepted record shapes (one JSON object per line, mixing allowed): //! -//! - `{"text": "...", "title"?: "..."}`: used as is. -//! - `{"prompt": "...", "completion": "..."}`: `title` = prompt, `text` = completion. -//! - `{"messages": [{"role", "content"}, ...]}`: the last user turn becomes -//! `title`, the last assistant turn `text`; a system turn (the same in every -//! record) becomes the job's `system_prompt`. +//! - `{"text": "...", "title"?: "..."}`: kept as `{title, text}`; the worker +//! trains it as one `user: "Brief me on: "` / `assistant: <text>` +//! exchange, loss on the whole exchange (as before). +//! - `{"prompt": "...", "completion": "..."}`: a one-exchange conversation, +//! the prompt verbatim as the user turn. +//! - `{"messages": [{"role", "content"}, ...]}`: kept whole, every turn in +//! order. The worker renders it with the base model's chat template and +//! computes the loss on the assistant turns only. A system turn the same in +//! every record also becomes the job's `system_prompt` (the adapter's +//! default for chat); records without one get that prompt at train time. //! -//! The worker formats each record as one chat exchange -//! (`system`, `user: "Brief me on: <title>"`, `assistant: <text>`), so -//! multi-turn conversations train on their final exchange only. +//! Conversations need a runner that reports the `messages` dataset format +//! (Infernet with peft `formats: ["text", "messages"]`), so a corpus holding +//! any requires [`crate::text::MESSAGES_FORMAT_TAG`] of the worker. use anyhow::{Result, bail}; use serde::Serialize; @@ -69,15 +74,18 @@ pub fn normalize_dataset(bytes: &[u8]) -> Result<(Vec<u8>, DatasetReport)> { if !v.is_object() { bail!("line {n}: each line must be a JSON object"); } - let (title, body) = if let Some(t) = str_field(&v, "text") { + let (record, bytes) = if let Some(t) = str_field(&v, "text") { report.text += 1; - (str_field(&v, "title").unwrap_or("").to_string(), t.to_string()) + let title = str_field(&v, "title").unwrap_or(""); + (json!({"title": title, "text": t}), title.len() + t.len()) } else if let (Some(p), Some(c)) = (str_field(&v, "prompt"), str_field(&v, "completion")) { + // The prompt verbatim as the user turn (not a "Brief me on:" title). report.prompt_completion += 1; - (p.to_string(), c.to_string()) + (json!({"messages": [{"role": "user", "content": p}, {"role": "assistant", "content": c}]}), p.len() + c.len()) } else if let Some(msgs) = v.get("messages").and_then(Value::as_array) { - let mut user = None; - let mut assistant = None; + let mut turns = Vec::with_capacity(msgs.len()); + let mut bytes = 0; + let mut learns = false; for m in msgs { let role = m.get("role").and_then(Value::as_str).unwrap_or(""); let content = m.get("content").and_then(Value::as_str); @@ -90,27 +98,31 @@ pub fn normalize_dataset(bytes: &[u8]) -> Result<(Vec<u8>, DatasetReport)> { Some(s) if s != content => system_conflict = true, _ => {} }, - "user" => user = Some(content), - "assistant" => assistant = Some(content), + "user" => {} + "assistant" => learns |= !content.trim().is_empty(), other => bail!("line {n}: message role must be system, user or assistant, got {other:?}"), } + bytes += content.len(); + turns.push(json!({"role": role, "content": content})); } - let Some(a) = assistant.filter(|a| !a.trim().is_empty()) else { + if !learns { bail!("line {n}: messages need an assistant turn to learn from"); - }; + } report.messages += 1; - (user.unwrap_or("").to_string(), a.to_string()) + // Every turn, in order: the worker trains the whole conversation + // (loss on the assistant turns). + (json!({"messages": turns}), bytes) } else { bail!("line {n}: expected {{\"text\"}}, {{\"prompt\",\"completion\"}} or {{\"messages\"}}"); }; - if title.len() + body.len() > MAX_RECORD_BYTES { + if bytes > MAX_RECORD_BYTES { bail!("line {n}: record is over {} KiB", MAX_RECORD_BYTES >> 10); } report.records += 1; if report.records > MAX_RECORDS { bail!("dataset has more than {MAX_RECORDS} records"); } - serde_json::to_writer(&mut out, &json!({"title": title, "text": body}))?; + serde_json::to_writer(&mut out, &record)?; out.push(b'\n'); } if report.records < MIN_RECORDS { @@ -247,8 +259,48 @@ mod tests { assert_eq!(r.system_prompt.as_deref(), Some("be brief")); let lines: Vec<Value> = std::str::from_utf8(&out).unwrap().lines().map(|l| serde_json::from_str(l).unwrap()).collect(); assert_eq!(lines[0], json!({"title": "A", "text": "a body"})); - assert_eq!(lines[1], json!({"title": "Q?", "text": "Ans."})); - assert_eq!(lines[2], json!({"title": "hi", "text": "hello"})); + assert_eq!(lines[1], json!({"messages": [{"role": "user", "content": "Q?"}, {"role": "assistant", "content": "Ans."}]})); + assert_eq!( + lines[2], + json!({"messages": [{"role": "system", "content": "be brief"}, {"role": "user", "content": "hi"}, {"role": "assistant", "content": "hello"}]}) + ); + assert!(crate::text::corpus_has_conversations(&out)); + } + + #[test] + fn multi_turn_chats_keep_every_turn_in_order() { + let chat = json!({"messages": [ + {"role": "system", "content": "You are a pricing coach."}, + {"role": "user", "content": "I sell a SaaS."}, + {"role": "assistant", "content": "What does it replace?"}, + {"role": "user", "content": "Spreadsheets."}, + {"role": "assistant", "content": "Price against the hours it saves."}, + ]}); + let src = format!("{chat}\n{chat}\n"); + let (out, r) = normalize_dataset(src.as_bytes()).unwrap(); + assert_eq!((r.records, r.messages), (2, 2)); + let first: Value = serde_json::from_str(std::str::from_utf8(&out).unwrap().lines().next().unwrap()).unwrap(); + // Nothing dropped, nothing reordered: both exchanges are there. + assert_eq!(first, chat); + assert_eq!(r.system_prompt.as_deref(), Some("You are a pricing coach.")); + // A record whose only assistant turn is empty teaches nothing. + let empty = b"{\"messages\":[{\"role\":\"user\",\"content\":\"q\"},{\"role\":\"assistant\",\"content\":\" \"}]}\n{\"text\":\"x\"}\n"; + assert!(normalize_dataset(empty).unwrap_err().to_string().contains("assistant")); + // Extra fields on a message are not carried to the worker. + let extra = b"{\"messages\":[{\"role\":\"user\",\"content\":\"q\",\"name\":\"n\"},{\"role\":\"assistant\",\"content\":\"a\"}]}\n{\"text\":\"x\"}\n"; + let (out, _) = normalize_dataset(extra).unwrap(); + assert!(!std::str::from_utf8(&out).unwrap().contains("\"name\"")); + // Text-only corpora stay {title, text} and need no format tag. + let (out, _) = normalize_dataset(b"{\"text\":\"a\"}\n{\"text\":\"b\",\"title\":\"t\"}\n").unwrap(); + assert!(!crate::text::corpus_has_conversations(&out)); + } + + #[test] + fn a_long_conversation_counts_toward_the_record_cap() { + let turn = "a".repeat(MAX_RECORD_BYTES / 4); + let msgs: Vec<Value> = (0..5).map(|i| json!({"role": if i % 2 == 0 { "user" } else { "assistant" }, "content": turn})).collect(); + let src = format!("{}\n{{\"text\":\"x\"}}\n", json!({"messages": msgs})); + assert!(normalize_dataset(src.as_bytes()).unwrap_err().to_string().contains("KiB")); } #[test] @@ -299,10 +351,15 @@ mod tests { // The same spec validation the worker runs. let spec = crate::text::TextTrainSpec { job: job.clone(), - dataset_url: "http://127.0.0.1:1/d".into(), + dataset_url: String::new(), dataset_hash: format!("sha256:{}", "a".repeat(64)), - artifact_upload_url: "http://127.0.0.1:1/u".into(), + artifact_upload_url: String::new(), events_url: None, + corpus_hash: Some(c0mpute_proto::Hash::of(b"corpus").to_hex()), + corpus_size: None, + corpus_pieces: None, + adapter_pieces: true, + run_id: Some("r1".into()), }; spec.validate().unwrap(); let o2 = LoraOptions { system_prompt: Some("flag".into()), ..o.clone() }; diff --git a/node/crates/c0mpute-core/tests/fixtures/fake-infernet-text b/node/crates/c0mpute-core/tests/fixtures/fake-infernet-text new file mode 100755 index 00000000..d83d4ea9 --- /dev/null +++ b/node/crates/c0mpute-core/tests/fixtures/fake-infernet-text @@ -0,0 +1,69 @@ +#!/usr/bin/env python3 +"""Stand-in for the `infernet` CLI's peft text commands in c0mpute tests. + +Same command lines and JSON shapes as Infernet's peft backend +(`model runtime status --peft --json`, `model train --json --job <job.json>`), +no PyTorch. The "adapter" is deterministic bytes derived from the corpus the +worker handed over (a few MiB, so it travels as several pieces), so the +buyer's hash checks are real. Like the real runner it reads only local paths +and refuses a corpus whose sha256 differs from the job's. +""" +import hashlib, json, os, sys, time + +args = sys.argv[1:] + +if args[:3] == ["model", "runtime", "status"]: + print(json.dumps({"installed": True, "cuda": False, "torch": "fake", "transformers": "fake", "peft": "fake", + "formats": ["text", "messages"]})) + sys.exit(0) + +def flag(name): + return args[args.index(name) + 1] if name in args else None + +if args[:2] == ["model", "generate"]: + if "--input" not in args: + # What Infernet >= 0.1.80 prints for a bare call (c0mpute probes it). + print("usage: infernet model generate --adapter <dir> | --input <request.json>", file=sys.stderr) + sys.exit(2) + req = json.load(open(flag("--input"))) + weights = open(os.path.join(req["adapter_dir"], "adapter_model.safetensors"), "rb").read() + turns = " | ".join(f"{m['role']}={m['content']}" for m in req["messages"]) + print(json.dumps({"v": 1, "type": "answer", "adapter": "sha256:" + hashlib.sha256(weights).hexdigest(), + "text": f"[{req['base_model']}] {turns}", "tokens": 7, "seconds": 0.1})) + sys.exit(0) + +def emit(**ev): + print(json.dumps({"v": 1, **ev}), flush=True) + +if args[:2] == ["model", "train"]: + job_path = flag("--job") + base = os.path.dirname(os.path.abspath(job_path)) + job = json.load(open(job_path)) + url = job["dataset"]["url"] + if "://" in url: + emit(type="error", message="dataset.url must be a local path"); emit(type="done", reason="error"); sys.exit(1) + raw = open(os.path.join(base, url), "rb").read() + digest = hashlib.sha256(raw).hexdigest() + if digest != job["dataset"].get("sha256", digest): + emit(type="error", message="dataset sha256 mismatch"); emit(type="done", reason="error"); sys.exit(1) + docs = [json.loads(l) for l in raw.decode().splitlines() if l.strip()] + turns = sum(len(d.get("messages", [])) for d in docs) + emit(type="meta", base_model=job["base_model"], device="cpu", total_steps=3, documents=len(docs), conversation_turns=turns) + for s in (1, 2, 3): + emit(type="step", step=s, total_steps=3, loss=round(2.0 / s, 3), throughput=10.0) + time.sleep(0.3) + out = os.path.join(base, job["output"]["destination_url"]) + os.makedirs(out, exist_ok=True) + seed = hashlib.sha256(raw).digest() + weights = (seed * (2_621_440 // len(seed) + 1))[:2_621_440] + open(os.path.join(out, "adapter_model.safetensors"), "wb").write(weights) + open(os.path.join(out, "adapter_config.json"), "w").write(json.dumps({"r": job["hyperparams"]["lora_r"]})) + open(os.path.join(out, "manifest.json"), "w").write(json.dumps({ + "base_model": job["base_model"], "dataset_sha256": digest, "documents": len(docs), + "conversation_turns": turns, "system_prompt": job.get("system_prompt")})) + emit(type="checkpoint", step=3, model="sha256:" + hashlib.sha256(weights).hexdigest(), size_bytes=len(weights)) + emit(type="done", reason="complete") + sys.exit(0) + +print("fake-infernet-text: unsupported " + " ".join(args), file=sys.stderr) +sys.exit(2) diff --git a/node/crates/c0mpute-core/tests/inference_e2e.rs b/node/crates/c0mpute-core/tests/inference_e2e.rs new file mode 100644 index 00000000..54e91ff7 --- /dev/null +++ b/node/crates/c0mpute-core/tests/inference_e2e.rs @@ -0,0 +1,146 @@ +//! End-to-end: `infernet.inference` peer to peer on loopback libp2p. +//! +//! A worker runs the real inference runner and bid filter in front of an +//! OpenAI-compatible runtime; a buyer auctions chat turns through the jobs +//! bridge (what `c0mpute chat --p2p` does) and gets answers fetched from the +//! worker and checked against its receipt. No c0mpute.com, no `infernet` +//! binary. The runtime is a stand-in on loopback, unless +//! `C0MPUTE_INFERENCE_LIVE_URL` names a real one (e.g. Ollama at +//! http://127.0.0.1:11434), in which case that one answers. + +use std::sync::Arc; +use std::time::Duration; + +use axum::routing::{get, post}; +use axum::{Json, Router}; +use c0mpute_core::inference; +use c0mpute_core::jobs_gateway::JobsBridge; +use c0mpute_core::pricing::PricingConfig; +use c0mpute_core::dispatch; +use c0mpute_net::{ChunkSource, Libp2pNetwork, NetworkConfig}; +use c0mpute_proto::Hash; +use c0mpute_store::ChunkStore; +use libp2p::multiaddr::Protocol; +use serde_json::{Value, json}; +use tempfile::tempdir; + +struct StoreSource(ChunkStore); + +#[async_trait::async_trait] +impl ChunkSource for StoreSource { + async fn read_chunk(&self, hash: &Hash) -> anyhow::Result<Vec<u8>> { + self.0.get(hash).await + } +} + +async fn node(dir: &std::path::Path, bootstrap: Vec<libp2p::Multiaddr>) -> (Arc<Libp2pNetwork>, ChunkStore) { + let store = ChunkStore::open(dir.join("store")).await.unwrap(); + let net = Libp2pNetwork::spawn( + NetworkConfig::for_dir(dir.join("id")) + .with_listen(vec!["/ip4/127.0.0.1/tcp/0".parse().unwrap()]) + .with_local_source(Arc::new(StoreSource(store.clone()))) + .with_bootstrap(bootstrap), + ) + .await + .unwrap(); + (Arc::new(net), store) +} + +async fn wait_for_listen(net: &Libp2pNetwork) -> libp2p::Multiaddr { + for _ in 0..50 { + if let Ok(addrs) = net.listen_addrs().await + && let Some(a) = addrs.into_iter().next() + { + return a; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!("no listen addr after 2.5s"); +} + +/// An OpenAI-compatible runtime that echoes the whole conversation it got. +async fn fake_runtime() -> String { + let app = Router::new() + .route("/v1/models", get(|| async { Json(json!({"object": "list", "data": [{"id": "tiny:1b"}]})) })) + .route( + "/v1/chat/completions", + post(|Json(b): Json<Value>| async move { + let turns: Vec<String> = b["messages"].as_array().unwrap().iter().map(|m| format!("{}={}", m["role"].as_str().unwrap(), m["content"].as_str().unwrap())).collect(); + Json(json!({"choices": [{"message": {"role": "assistant", "content": format!("[{}] {}", b["model"].as_str().unwrap(), turns.join(" | "))}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 5, "completion_tokens": 5}})) + }), + ); + let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = l.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(l, app).await.unwrap() }); + format!("http://{addr}") +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn chat_turns_are_answered_peer_to_peer() { + let _ = tracing_subscriber::fmt().with_env_filter("c0mpute_core=info,c0mpute_net=warn").with_test_writer().try_init(); + let live = std::env::var("C0MPUTE_INFERENCE_LIVE_URL").ok().filter(|s| !s.is_empty()); + let base = match &live { + Some(u) => u.trim_end_matches('/').to_string(), + None => fake_runtime().await, + }; + let tmp = tempdir().unwrap(); + + // ── worker: refresh loop, runner, bid filter ─────────────────────── + tokio::spawn(inference::refresh_loop(base.clone())); + let mut models = Vec::new(); + for _ in 0..50 { + models = inference::ready_models(); + if !models.is_empty() { + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + assert!(!models.is_empty(), "the runtime at {base} lists no chat models"); + let model = models[0].clone(); + let (worker, worker_store) = node(&tmp.path().join("w"), vec![]).await; + let worker_addr = wait_for_listen(&worker).await.with(Protocol::P2p(worker.peer_id())); + let load = inference::worker_load(); + let runner = inference::spawn_inference_runner(worker.clone(), worker_store, base.clone(), load.clone()); + let mut tags = vec![inference::ROLE_TAG.to_string(), "c0mpute:cpu".to_string()]; + tags.extend(models.iter().map(|m| inference::model_tag(m))); + dispatch::run_worker_subscriber_with_load(worker.clone(), inference::WORKLOAD.to_string(), tags, Some(runner), Some(inference::offer_filter()), Some(PricingConfig::default()), Some(load)); + + // ── buyer ────────────────────────────────────────────────────────── + let (buyer, buyer_store) = node(&tmp.path().join("b"), vec![worker_addr]).await; + let bridge = JobsBridge::new(buyer.clone(), buyer_store, "test-token-0123456789abcdef".into(), 0.0, vec![]).unwrap().with_bid_window(Duration::from_secs(2)); + tokio::time::sleep(Duration::from_secs(2)).await; + + // A multi-turn conversation, any model. + let spec = json!({"v": 1, "max_tokens": 32, "temperature": 0.0, "messages": [ + {"role": "system", "content": "Answer in one word."}, + {"role": "user", "content": "Name a primary color."}, + {"role": "assistant", "content": "Red."}, + {"role": "user", "content": "Another one?"}, + ]}); + let out = bridge.submit_once(inference::WORKLOAD, spec, Duration::from_secs(240)).await.expect("answered"); + assert_eq!(out["worker"], worker.peer_id().to_base58()); + let r = &out["result"]; + assert!(inference::valid_result(r), "{r}"); + assert_eq!(r["ok"], true, "{r}"); + assert_eq!(r["model"], model.as_str()); + let text = r["text"].as_str().unwrap(); + assert!(!text.trim().is_empty()); + if live.is_none() { + // The worker forwarded every turn, in order, to the runtime. + assert_eq!(text, "[tiny:1b] system=Answer in one word. | user=Name a primary color. | assistant=Red. | user=Another one?"); + } else { + eprintln!("live answer from {model}: {text}"); + } + + // Asking for the model by name works; a model nobody serves gets no bid. + let named = json!({"v": 1, "model": model, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 8}); + let out = bridge.submit_once(inference::WORKLOAD, named, Duration::from_secs(240)).await.expect("named model"); + assert_eq!(out["result"]["model"], model.as_str()); + let missing = json!({"v": 1, "model": "not-served:9b", "messages": [{"role": "user", "content": "hi"}]}); + let err = bridge.submit_once(inference::WORKLOAD, missing, Duration::from_secs(8)).await.unwrap_err().to_string(); + assert!(err.contains("no_bids") || err.contains("timeout"), "{err}"); + // An invalid spec never leaves the buyer. + let bad = json!({"v": 1, "messages": [{"role": "tool", "content": "x"}]}); + assert!(bridge.submit_once(inference::WORKLOAD, bad, Duration::from_secs(5)).await.is_err()); +} diff --git a/node/crates/c0mpute-core/tests/text_train_e2e.rs b/node/crates/c0mpute-core/tests/text_train_e2e.rs new file mode 100644 index 00000000..0ad8df1e --- /dev/null +++ b/node/crates/c0mpute-core/tests/text_train_e2e.rs @@ -0,0 +1,185 @@ +//! End-to-end: `infernet.text.train` peer to peer on loopback libp2p. +//! +//! A worker (`text::serve`) and a buyer (`text::train`) with no HTTP between +//! them: the buyer stores a multi-MiB corpus (with multi-turn conversations) +//! as pieces, the worker pulls them over libp2p, publishes training events on +//! the job topic, stores the adapter as pieces behind a manifest its receipt +//! names, and the buyer fetches and verifies it. A fake `infernet` +//! (tests/fixtures/fake-infernet-text, no PyTorch) stands in for the peft +//! runner; it writes a deterministic adapter derived from the corpus it was +//! handed, so every hash check is real. Skips where python3 is missing. + +use std::path::PathBuf; +use std::time::Duration; + +use c0mpute_core::temporal::node::{NetOptions, spawn_node_with_store}; +use c0mpute_core::text::{self, TrainRequest}; +use c0mpute_core::train::{LoraOptions, normalize_dataset}; +use c0mpute_store::ChunkStore; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; +use tempfile::tempdir; + +fn fake_infernet() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/fake-infernet-text") +} + +fn python3() -> bool { + std::process::Command::new("python3").arg("--version").output().is_ok_and(|o| o.status.success()) +} + +fn net(dir: &std::path::Path, bootstrap: Vec<String>) -> NetOptions { + NetOptions { identity_dir: dir.to_path_buf(), listen: vec!["/ip4/127.0.0.1/tcp/0".into()], bootstrap, default_bootstrap: false } +} + +/// About 3 MiB of records: text, prompt/completion and multi-turn chats. +fn dataset() -> Vec<u8> { + let mut out = String::new(); + let mut i = 0u64; + while out.len() < 3 * 1024 * 1024 { + let filler: String = (0..40).map(|k| format!("w{}", (i * 31 + k) % 997)).collect::<Vec<_>>().join(" "); + let rec = match i % 3 { + 0 => json!({"title": format!("Topic {i}"), "text": filler}), + 1 => json!({"prompt": format!("Q{i}?"), "completion": filler}), + _ => json!({"messages": [ + {"role": "system", "content": "You are a pricing coach."}, + {"role": "user", "content": format!("I sell product {i}.")}, + {"role": "assistant", "content": "What does it replace?"}, + {"role": "user", "content": "Spreadsheets."}, + {"role": "assistant", "content": filler}, + ]}), + }; + out.push_str(&rec.to_string()); + out.push('\n'); + i += 1; + } + out.into_bytes() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn trains_peer_to_peer_with_pieces_and_conversations() { + if !python3() { + eprintln!("skipping: python3 is not installed (the fake infernet is a python script)"); + return; + } + let _ = tracing_subscriber::fmt().with_env_filter("c0mpute_core=info,c0mpute_net=warn").with_test_writer().try_init(); + let tmp = tempdir().unwrap(); + + // ── worker ───────────────────────────────────────────────────────── + let w = text::serve(&net(&tmp.path().join("w/id"), vec![]), fake_infernet(), tmp.path().join("w/cache")).await.expect("worker serves"); + assert!(w.tags.contains(&"infernet:text:train".to_string())); + assert!(w.tags.contains(&text::MESSAGES_FORMAT_TAG.to_string()), "the runner reports the messages format: {:?}", w.tags); + assert!(w.infer, "the fake answers chats (model generate)"); + assert!(w.tags.contains(&"infernet:text:infer".to_string())); + let wid = w.net.peer_id().to_base58(); + let waddr = format!("{}/p2p/{wid}", w.net.listen_addrs().await.unwrap()[0]); + + // ── buyer: what `c0mpute train lora` does ────────────────────────── + let raw = dataset(); + let (corpus, report) = normalize_dataset(&raw).unwrap(); + assert!(report.messages > 0 && report.text > 0 && report.prompt_completion > 0); + assert!(corpus.len() > 2 * 1024 * 1024, "several pieces"); + let opts = LoraOptions { base_model: "HuggingFaceTB/SmolLM2-135M-Instruct".into(), name: "coach".into(), ..Default::default() }; + let job = opts.job(&report); + let stage = tmp.path().join("stage"); + std::fs::create_dir_all(&stage).unwrap(); + std::fs::write(stage.join("corpus.jsonl"), &corpus).unwrap(); + std::fs::write(stage.join("job.json"), serde_json::to_vec(&job).unwrap()).unwrap(); + + let store = ChunkStore::open(tmp.path().join("b/store")).await.unwrap(); + let buyer = spawn_node_with_store(&net(&tmp.path().join("b/id"), vec![waddr]), store.clone()).await.unwrap(); + tokio::time::sleep(Duration::from_secs(2)).await; + + let req = TrainRequest { + job_path: stage.join("job.json"), + dataset_path: Some(stage.join("corpus.jsonl")), + output: tmp.path().join("adapter"), + workers: vec![], + max_price_usd: 1.0, + gpu: false, + min_ram_gb: None, + legacy_http: false, + public_url: None, + bind: "127.0.0.1:0".parse().unwrap(), + timeout_s: 120, + retries: 1, + bid_window: Duration::from_secs(2), + }; + let mut events: Vec<Value> = Vec::new(); + let out = text::train(buyer.clone(), &store, &req, |ev| events.push(ev.clone())).await.expect("trains"); + + // Peer to peer both ways. + assert_eq!(out["transport"], "pieces", "{out}"); + assert_eq!(out["receipt"]["worker_peer_id"], wid); + let manifest_hash = out["receipt"]["output_hash"].as_str().unwrap(); + assert!(!manifest_hash.starts_with("sha256:"), "the receipt names the adapter manifest"); + + // The adapter is exactly what the trainer derived from the corpus the + // buyer sent: the corpus arrived intact. + let seed = Sha256::digest(&corpus); + let want: Vec<u8> = seed.iter().copied().cycle().take(2_621_440).collect(); + let got = std::fs::read(tmp.path().join("adapter/adapter_model.safetensors")).unwrap(); + assert_eq!(got.len(), want.len()); + assert!(got == want, "adapter bytes differ"); + assert_eq!(out["adapter_hash"], format!("sha256:{}", hex::encode(Sha256::digest(&want)))); + let m: Value = serde_json::from_slice(&std::fs::read(tmp.path().join("adapter/manifest.json")).unwrap()).unwrap(); + assert_eq!(m["dataset_sha256"], hex::encode(seed)); + assert_eq!(m["system_prompt"], "You are a pricing coach."); + // Every conversation reached the trainer whole: five turns per chat, + // two per prompt/completion pair. + assert_eq!(m["conversation_turns"].as_u64().unwrap() as usize, report.messages * 5 + report.prompt_completion * 2); + assert!(std::fs::read(tmp.path().join("adapter/adapter_config.json")).is_ok()); + + // Progress came over gossip. + let kinds: Vec<&str> = events.iter().filter_map(|e| e["type"].as_str()).collect(); + assert!(kinds.contains(&"meta"), "{kinds:?}"); + assert!(kinds.contains(&"checkpoint"), "{kinds:?}"); + assert_eq!(kinds.last(), Some(&"done"), "{kinds:?}"); + + // The worker cleaned its job dir (corpus and adapter now live as pieces). + let left = std::fs::read_dir(tmp.path().join("w/cache/text/jobs")).map(|d| d.count()).unwrap_or(0); + assert_eq!(left, 0, "worker job dir removed"); + + // A tampered manifest hash is refused (nothing is written). + let bogus = c0mpute_proto::Hash::of(b"not a manifest").to_hex(); + let err = text::fetch_adapter_pieces(&buyer, Some(w.net.peer_id()), &bogus, &tmp.path().join("bad")).await; + assert!(err.is_err()); + assert!(!tmp.path().join("bad").exists()); + + // ── then chat with the adapter, also peer to peer ────────────────── + // (`c0mpute infernet text chat`: adapter out as pieces, answer back as a chunk) + let chat = text::ChatRequest { + adapter_dir: tmp.path().join("adapter"), + question: "How should I price it?".into(), + search: None, + max_new_tokens: 64, + workers: vec![], + max_price_usd: 1.0, + legacy_http: false, + public_url: None, + bind: "127.0.0.1:0".parse().unwrap(), + timeout_s: 60, + bid_window: Duration::from_secs(2), + }; + let ans = text::chat(buyer.clone(), &store, &chat).await.expect("chat answered"); + assert_eq!(ans["text"], "[HuggingFaceTB/SmolLM2-135M-Instruct] system=You are a pricing coach. | user=How should I price it?", "{ans}"); + assert_eq!(ans["adapter_hash"], out["adapter_hash"]); + assert!(!ans["receipt"]["output_hash"].as_str().unwrap().starts_with("sha256:"), "answer came back as a chunk"); + // The worker cached the adapter it fetched as pieces. + let cached = tmp.path().join("w/cache/text/adapters").join(out["adapter_hash"].as_str().unwrap().trim_start_matches("sha256:")); + assert_eq!(std::fs::read(cached.join("adapter_model.safetensors")).unwrap(), want); +} + +/// A corpus with conversations is never placed on a worker whose runner +/// cannot train them (it would silently drop them). +#[test] +fn conversations_need_a_runner_that_reports_messages() { + let chat = br#"{"messages":[{"role":"user","content":"a"},{"role":"assistant","content":"b"}]}"#; + let mut corpus = chat.to_vec(); + corpus.push(b'\n'); + let tags = text::required_tags(&corpus); + let old_runner = text::worker_tags(&json!({"installed": true, "cuda": false}), 8 * 1024); + assert!(!tags.iter().all(|t| old_runner.contains(t)), "an old runner lacks {}", text::MESSAGES_FORMAT_TAG); + let new_runner = text::worker_tags(&json!({"installed": true, "cuda": false, "formats": ["text", "messages"]}), 8 * 1024); + assert!(tags.iter().all(|t| new_runner.contains(t))); +} diff --git a/node/crates/c0mpute-core/tests/workflow_e2e.rs b/node/crates/c0mpute-core/tests/workflow_e2e.rs index 2d4bd188..89989baf 100644 --- a/node/crates/c0mpute-core/tests/workflow_e2e.rs +++ b/node/crates/c0mpute-core/tests/workflow_e2e.rs @@ -113,7 +113,8 @@ async fn transform_step_round_trips_through_the_bridge() { "crawl.fetch", "ocr.extract", "embed.text", - "blender.render-frame" + "blender.render-frame", + "infernet.inference" ]) ); diff --git a/node/crates/c0mpute-proto/src/lib.rs b/node/crates/c0mpute-proto/src/lib.rs index ed38a850..4d153e11 100644 --- a/node/crates/c0mpute-proto/src/lib.rs +++ b/node/crates/c0mpute-proto/src/lib.rs @@ -219,6 +219,10 @@ pub enum Role { /// Blender (Cycles on CPU, or on a GPU Blender finds). On by default; /// advertised only where `blender --version` works. Render, + /// `infernet.inference`: chat completions from the box's own + /// OpenAI-compatible runtime (Ollama by default, or llama-server). On by + /// default; advertised only while the runtime lists chat models. + Inference, } #[cfg(test)] diff --git a/plugins/infernet/module.toml b/plugins/infernet/module.toml index 2e2ed7d9..d585bf14 100644 --- a/plugins/infernet/module.toml +++ b/plugins/infernet/module.toml @@ -1,7 +1,19 @@ # c0mpute plugin manifest. See dips/0006-module-model.md. # -# Source code lives upstream at github.com/infernetprotocol/infernet-protocol, NOT -# this repo. This directory is a marketplace wrapper only. +# Two halves: +# +# - `infernet.inference` (chat) is NATIVE in the c0mpute binary since this +# release: buyers run `c0mpute chat --p2p "..."` (the default when not signed +# in to c0mpute.com) or POST it to `c0mpute jobs serve`; workers answer from +# their own OpenAI-compatible runtime (Ollama at 127.0.0.1:11434 by default, +# or llama-server etc. via C0MPUTE_INFERENCE_URL; `off` disables), role +# `inference`, on by default, advertised only while the runtime lists chat +# models (tags infernet:inference + infernet:inference:model:<id>). Answers +# come back as content-addressed chunks checked against the receipt. Neither +# side needs the infernet binary or c0mpute.com (node/crates/c0mpute-core/src/inference.rs). +# - The temporal and text (LoRA) workloads below still run Infernet's own +# runtimes on the worker, so those workers need the infernet CLI (installed by +# install.sh). Source: github.com/infernetprotocol/infernet-protocol. # # `version` is this wrapper's, kept in lockstep with c0mpute by # scripts/bump-version.sh. The installed infernet CLI's own version is what @@ -13,9 +25,8 @@ name = "Infernet Protocol — AI inference" version = "0.2.79" kind = "workload" category = "ai" -status = "partial" -status_note = "Chat runs through the separate infernet CLI, not the c0mpute binary." -description = "Decentralized AI inference workloads on c0mpute. Runs models on consumer/prosumer GPUs with model-hash and runtime-image verification. Also trains and serves temporal policies (CfC/LTC liquid networks, GRU baseline) and fine-tunes language models (LoRA) with live progress, on CPU or GPU." +status = "available" +description = "Chat with open models peer to peer from the c0mpute binary itself (`c0mpute chat --p2p`): workers answer from their own Ollama or llama.cpp server, no account or extra CLI. Also trains and serves temporal policies (CfC/LTC liquid networks, GRU baseline) and fine-tunes language models (LoRA) with live progress, on CPU or GPU." author = "Infernet Protocol" license = "MIT" homepage = "https://infernetprotocol.com" @@ -23,13 +34,15 @@ source = "https://github.com/infernetprotocol/infernet-protocol" keywords = ["ai", "llm", "inference", "gpu", "temporal-policy", "liquid-neural-network", "cfc", "lora", "fine-tuning"] [module.requirements] -c0mpute = ">=0.1.0" +c0mpute = ">=NEXT" os = ["linux", "darwin"] arch = ["x86_64", "aarch64"] capabilities = ["gpu:nvidia?", "model-runtime"] [module.workloads] -"infernet.inference" = { command = "chat", validation = "schema_and_spotcheck" } +# Native (no infernet binary): `c0mpute chat --p2p`; workers serve their local +# OpenAI-compatible runtime. Receipts carry the answer chunk's content hash. +"infernet.inference" = { command = "c0mpute chat --p2p", runs = "builtin", validation = "answer_chunk_hash" } # Temporal policies (liquid-agent/v1). Executed by Infernet >= 0.1.66 # (`infernet model train --job` / `infernet model serve --stdio`, which run # the infernet-liquid runtime the operator installs with @@ -67,6 +80,11 @@ runtime = "infernet:text:runtime:peft" devices = ["infernet:text:device:cpu", "infernet:text:device:cuda"] ram = "c0mpute:ram:<n>gb+" +[module.capabilities.inference] +# Tags a worker serving chat advertises; buyers naming a model require its tag. +role = "infernet:inference" +model = "infernet:inference:model:<id>" + [module.surfaces] -cli = "infernet" +cli = "c0mpute chat --p2p" # web = "c0mpute.com/infernet" # planned diff --git a/plugins/train/module.toml b/plugins/train/module.toml index 647ca9ce..57536aa4 100644 --- a/plugins/train/module.toml +++ b/plugins/train/module.toml @@ -8,20 +8,32 @@ # 1. checks the dataset: UTF-8 JSONL, one object per line as {text, title?}, # {prompt, completion} or {messages: [{role, content}]}; at least 2 # records, at most 1,000,000 records, 1 MiB per record, 512 MiB in all; -# 2. converts it to the {title, text} corpus the worker's peft runner reads -# (prompt -> title, completion -> text; for messages the last user and -# assistant turns, and a shared system turn becomes the system prompt); +# 2. converts it to the corpus the worker's peft runner reads: {title, text} +# stays a "Brief me on: <title>" exchange; prompt/completion and messages +# become whole conversations ({messages: [...]}, every turn in order) that +# the runner renders with the base model's chat template and trains with +# the loss on the assistant turns only; # 3. writes the @infernetprotocol/training job (kind "lora", base_model, # hyperparams lr / epochs / lora_r / seq_len / batch_size / max_steps); -# 4. auctions it through text::train, relays live progress, downloads the -# adapter and keeps it only if its sha256 matches the receipt; -# 5. prints how to use it: `c0mpute infernet text chat --adapter <out>`. +# 4. auctions it through text::train PEER TO PEER: the corpus goes out as +# content-addressed 1 MiB pieces the worker pulls over libp2p, progress +# comes back as gossip on the job topic, and the adapter comes back as +# pieces behind a manifest whose hash the receipt carries; the buyer keeps +# the adapter only if its sha256 matches. Nothing needs a reachable HTTP +# address (`--legacy-http` additionally serves workers from before the pieces transport over HTTP); +# 5. prints how to use it: `c0mpute infernet text chat --adapter <out>`, which +# is peer to peer the same way (adapter out as pieces, answer back as a chunk). # -# Worker side (unchanged): `c0mpute infernet text serve`, which needs Infernet -# >= 0.1.79 with the peft runtime (`infernet model runtime setup --peft --yes`: -# torch + transformers + peft in Infernet's venv). Training is fp32 LoRA on CPU -# or a CUDA GPU when torch sees one (buyers can require it with --gpu); no GPU -# is required, so keep base models small (0.5B-class trains on CPU in hours). +# Worker side: `c0mpute infernet text serve`. c0mpute moves the data and hands +# Infernet's trainer local file paths; the training itself needs Infernet with +# the peft runtime (`infernet model runtime setup --peft --yes`: torch + +# transformers + peft in Infernet's venv). A corpus with conversations goes +# only to workers whose runner reports the `messages` dataset format (tag +# infernet:text:format:messages, Infernet with infernetprotocol/infernet-protocol#161 +# "peft: train multi-turn conversations"); older runners would drop them. Training is +# fp32 LoRA on CPU or a CUDA GPU when torch sees one (buyers can require it with +# --gpu); no GPU is required, so keep base models small (0.5B-class trains on +# CPU in hours). # # No QLoRA: the peft runner loads the base model in fp32 with no 4-bit # quantization (no bitsandbytes), so `train.qlora` is not offered. @@ -32,22 +44,21 @@ name = "Fine-tune — LoRA training" version = "0.2.79" kind = "workload" category = "ai" -status = "partial" -status_note = "LoRA works end to end through infernet.text.train (`c0mpute train lora`). The buyer serves the corpus and receives the adapter over HTTP from its own machine, so the worker must reach it: same LAN, or pass --public-url to a forwarded https address. Each record trains as one 'Brief me on: <title>' exchange, so multi-turn chats keep only their last turn. No QLoRA (the peft runner trains fp32 LoRA)." +status = "available" builtin = true -description = "Fine-tune a LoRA adapter on a c0mpute worker from a JSONL dataset (text, prompt/completion or chat messages): live progress, adapter verified by hash, then chat with it." +description = "Fine-tune a LoRA adapter on a c0mpute worker from a JSONL dataset (text, prompt/completion or multi-turn chats, trained on every turn): dataset and adapter move peer to peer as verified pieces, live progress, then chat with it." author = "Profullstack" license = "MIT" keywords = ["train", "finetune", "lora", "peft", "adapter", "infernet"] [module.requirements] -c0mpute = ">=0.2.77" +c0mpute = ">=NEXT" os = ["linux", "darwin"] arch = ["x86_64", "aarch64"] -capabilities = ["bin:infernet", "infernet:text:runtime:peft", "gpu:cuda?"] +capabilities = ["bin:infernet", "infernet:text:runtime:peft", "infernet:text:format:messages?", "gpu:cuda?"] [module.workloads] -"train.lora" = { command = "lora", runs = "infernet.text.train", validation = "adapter_sha256_matches_receipt" } +"train.lora" = { command = "lora", runs = "infernet.text.train", validation = "adapter_sha256_matches_receipt_manifest" } [module.dispatch] mode = "subprocess"