Finish train + infernet: p2p dataset/adapter pieces, multi-turn chats, native p2p chat

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<String>,
@@ -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<String>, messages: &[Value]) -> Result<Value> {
+    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<Self> {
+        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<String>, 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<Value>) -> Result<()> {
+        let mut model = a.model.clone();
+        eprintln!("c0mpute chat (peer to peer) · {} · /model <id>, /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]

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=<secret> 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 <id>\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:<hash> --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=<secret> 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 <id>\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:<hash> --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.
@@ -1594,6 +1594,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);
@@ -4114,6 +4115,7 @@ fn parse_role(s: &str) -> Option<Role> {
         "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

node/crates/c0mpute-cli/src/text.rs

@@ -57,7 +57,10 @@ enum Sub {
         worker: Vec<String>,
         #[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<String>,
         #[arg(long, default_value = "0.0.0.0:0")]
@@ -91,7 +94,10 @@ enum Sub {
         gpu: bool,
         #[arg(long)]
         min_ram_gb: Option<u64>,
-        /// 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<String>,
         #[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;