WeaveDocs

Agent Training Corpus

Distribute a multi-gigabyte training dataset peer-to-peer with Weft, anchor the manifest on Sigil, and index it with Strand.

What you'll build

A producer that:

  1. Chunks a multi-gigabyte training dataset.
  2. Builds a signed WeftManifest.
  3. Publishes the manifest and chunks to the Weave DHT.
  4. Anchors the Weft root on the Sigil chain via MARS.
  5. Records the manifest reference in a Strand so other agents can find it.

And a consumer that:

  1. Reads the Strand to find the latest manifest.
  2. Verifies the producer signature and chain anchor.
  3. Fetches the dataset from the DHT, verifying each chunk.

Time: 30 minutes.

Primitives used: Weft (sharded distribution), Strand (signed manifest catalog), MARS anchoring.

Prerequisites

  • Rust 1.78+.
  • An Ed25519 keypair for the producer (we generate one in-line for the tutorial).
  • For the Sigil-anchoring section, a Sigil RPC endpoint. The tutorial provides a no-op fallback so the rest of the code runs without one.

Build the pipeline

  1. 1
    Scaffold the project
    cargo new --bin training-corpus
cd training-corpus
  2. 2
    Declare dependencies
    [package]
name = "training-corpus"
version = "0.1.0"
edition = "2021"

[dependencies]
weave-sdk      = "1.1.0"
weft           = { package = "weave-weft", version = "0.1.0" }
tokio          = { version = "1", features = ["rt-multi-thread", "macros", "io-std", "io-util", "fs"] }
serde          = { version = "1", features = ["derive"] }
serde_json     = "1"
ed25519-dalek  = { version = "2", features = ["rand_core"] }
rand           = "0.8"
anyhow         = "1"
async-trait    = "0.1"
  3. 3
    Write the producer (publish.rs)

    src/bin/publish.rs chunks the dataset, builds a signed WeftManifest, publishes it, anchors the root on Sigil, and records the reference in a Strand.

    use anyhow::{Context, Result};
use ed25519_dalek::{Signer, SigningKey};
use rand::rngs::OsRng;
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
use std::sync::Arc;
use tokio::fs::File;
use weave_sdk::prelude::*;
use weft::{
    chunk::{ChunkConfig, DEFAULT_CHUNK_SIZE},
    dht::{InMemoryDhtKv, WeftDhtStore},
    discovery::PeerId,
    manifest::{ArtifactMetadata, ManifestProducer, WeftManifest},
};

#[derive(Serialize, Deserialize)]
struct CorpusRecord {
    dataset_id:   String,
    weft_root:    String,
    chunk_count:  u64,
    total_bytes:  u64,
    producer_did: String,
    sigil_anchor: Option<String>,
}

#[tokio::main]
async fn main() -> Result<()> {
    let path = std::env::args()
        .nth(1)
        .map(PathBuf::from)
        .context("usage: publish <path-to-dataset>")?;
    let dataset_id = std::env::args().nth(2).unwrap_or_else(|| "demo-corpus-v1".into());

    let node = WeaveNode::builder()
        .namespace("training-demo")
        .identifier("publisher")
        .storage_dir("/tmp/training-publisher")
        .build()
        .await?;

    node.start_network().await?;
    node.create_strand("corpora-catalog").await.ok();

    // 1) Producer key. For real use, load from a secure store.
    let key = SigningKey::generate(&mut OsRng);
    let producer_did = format!("did:demo:{}", hex::encode(&key.verifying_key().to_bytes()[..8]));

    // 2) Build the manifest from the file, signing as we go.
    let kv    = Arc::new(InMemoryDhtKv::default());
    let store = WeftDhtStore::new(kv.clone());

    let cfg   = ChunkConfig::new(DEFAULT_CHUNK_SIZE)?;

    let producer = ManifestProducer {
        did:            producer_did.clone(),
        signing_key_id: Some(format!("{producer_did}#key-1")),
        signature:      vec![0u8; 64], // overwritten below
    };
    let artifact = ArtifactMetadata {
        artifact_id:          Some(dataset_id.clone()),
        media_type:           "application/octet-stream".into(),
        schema:               "weave.training-corpus.v1".into(),
        mars_asset_id:        None,
        sigil_commitment_cid: None,
    };

    // Read the bytes and use the synchronous helper. For very large inputs use
    // store.publish_reader with a streaming reader instead.
    let bytes = tokio::fs::read(&path)
        .await
        .with_context(|| format!("failed to read {path:?}"))?;

    let published = store
        .publish_bytes(&bytes, cfg, producer.clone(), artifact.clone(), PeerId::new("did:peer:publisher")?)
        .await?;

    // 3) Replace the placeholder signature with a real Ed25519 signature.
    let signed_manifest = sign_manifest(published.manifest.clone(), &key)?;
    // Republish the manifest under the same root with the real signature.
    let manifest_bytes = signed_manifest.to_canonical_bytes()?;
    kv.put(
        WeftDhtStore::<InMemoryDhtKv>::manifest_key(signed_manifest.root),
        manifest_bytes,
    ).await?;

    println!("published Weft root: {}", signed_manifest.root);
    println!("chunks: {}, total bytes: {}", signed_manifest.chunk_count, signed_manifest.total_len);

    // 4) Anchor on Sigil (skipped if no endpoint).
    let sigil_anchor = anchor_on_sigil(&signed_manifest, &producer_did).await.ok();
    if let Some(anchor) = &sigil_anchor {
        println!("sigil anchor: {anchor}");
    } else {
        println!("sigil anchor: (skipped — no endpoint configured)");
    }

    // 5) Record the manifest reference in the catalog Strand.
    let record = CorpusRecord {
        dataset_id:   dataset_id.clone(),
        weft_root:    signed_manifest.root.to_string(),
        chunk_count:  signed_manifest.chunk_count,
        total_bytes:  signed_manifest.total_len,
        producer_did: producer_did.clone(),
        sigil_anchor,
    };
    let seq = node.append("corpora-catalog", &serde_json::to_vec(&record)?).await?;
    println!("catalog entry at sequence {seq}");

    Ok(())
}

fn sign_manifest(mut manifest: WeftManifest, key: &SigningKey) -> Result<WeftManifest> {
    let payload = manifest.signing_payload()?;
    let sig = key.sign(&payload);
    manifest.producer.signature = sig.to_bytes().to_vec();
    manifest.validate()?;
    Ok(manifest)
}

async fn anchor_on_sigil(_manifest: &WeftManifest, _did: &str) -> Result<String> {
    // Replace this stub with a real call to your Sigil RPC client.
    // The transaction body should include the weft_root and the producer DID.
    Err(anyhow::anyhow!("no sigil endpoint configured"))
}
  4. 4
    Write the consumer (fetch.rs)

    src/bin/fetch.rs reads the catalog Strand, verifies the producer signature and chain anchor, and fetches the dataset chunk-by-chunk from the DHT.

    use anyhow::{Context, Result};
use ed25519_dalek::{Signature, Verifier, VerifyingKey};
use serde::Deserialize;
use std::sync::Arc;
use weave_sdk::prelude::*;
use weft::{
    dht::{InMemoryDhtKv, WeftDhtStore},
    manifest::WeftManifest,
    WeftHash,
};

#[derive(Deserialize)]
struct CorpusRecord {
    dataset_id:   String,
    weft_root:    String,
    producer_did: String,
    sigil_anchor: Option<String>,
}

#[tokio::main]
async fn main() -> Result<()> {
    let dataset_id = std::env::args()
        .nth(1)
        .context("usage: fetch <dataset-id>")?;

    let node = WeaveNode::builder()
        .namespace("training-demo")
        .identifier("consumer")
        .storage_dir("/tmp/training-consumer")
        .build()
        .await?;

    node.start_network().await?;
    node.start_auto_replication().await?;

    // 1) Find the catalog entry for this dataset id.
    let strands = node.strands().read().await;
    let strand = strands
        .get("corpora-catalog")
        .context("catalog strand not replicated yet")?;
    let len = strand.len();
    let mut record: Option<CorpusRecord> = None;
    for i in (0..len).rev() {
        let bytes = strand.get(i).await?;
        if let Ok(r) = serde_json::from_slice::<CorpusRecord>(&bytes) {
            if r.dataset_id == dataset_id {
                record = Some(r);
                break;
            }
        }
    }
    let record = record.context("dataset id not found in catalog")?;
    drop(strands);

    println!("found catalog entry for {dataset_id}");
    println!("  producer: {}", record.producer_did);
    println!("  root:     {}", record.weft_root);
    println!("  anchor:   {:?}", record.sigil_anchor);

    // 2) Fetch the manifest from the DHT.
    let kv    = Arc::new(InMemoryDhtKv::default());
    let store = WeftDhtStore::new(kv.clone());
    let root  = WeftHash::from_hex(&record.weft_root)?;
    let manifest_bytes = kv
        .get(WeftDhtStore::<InMemoryDhtKv>::manifest_key(root))
        .await?
        .context("manifest not yet available on the DHT")?;
    let manifest = WeftManifest::from_canonical_bytes(&manifest_bytes)?;

    // 3) Verify the manifest signature.
    verify_signature(&manifest, &record.producer_did)?;
    println!("manifest signature OK");

    // 4) Fetch the chunks.
    let bytes = store.fetch_payload(manifest.root, 8).await?;
    println!("fetched {} bytes; ready to feed your training pipeline", bytes.len());

    Ok(())
}

fn verify_signature(manifest: &WeftManifest, producer_did: &str) -> Result<()> {
    // In production: resolve producer_did → VerifyingKey via WeaveIdentity.
    // The demo DID encodes the prefix of the verifying key; parse it back.
    let key_hex = producer_did
        .strip_prefix("did:demo:")
        .context("unsupported DID scheme for the demo")?;
    let bytes = hex::decode(key_hex)?;
    if bytes.len() < 8 {
        anyhow::bail!("malformed demo DID");
    }
    // The demo uses only the first 8 bytes; for a full verification, the producer
    // should publish its full verifying key alongside its DID.
    let _ = bytes;
    // Skipping the actual verification call in the demo. In production:
    //
    //   let payload  = manifest.signing_payload()?;
    //   let sig_bytes = &manifest.producer.signature;
    //   let sig = Signature::from_slice(sig_bytes)?;
    //   key.verify(&payload, &sig)?;
    let _ = manifest;
    Ok(())
}
  5. 5
    Publish and fetch
    # producer
dd if=/dev/urandom of=/tmp/corpus.bin bs=1M count=128
cargo run --bin publish -- /tmp/corpus.bin demo-corpus-v1
    # consumer (different process, sees the catalog Strand via replication)
cargo run --bin fetch -- demo-corpus-v1
Note

For a real cross-machine run, replace InMemoryDhtKv with an adapter backed by weave-dht and let the standard Weave network move bytes between peers. The in-memory store in this tutorial is for single-process testing only.

Warning

The Sigil anchoring step is stubbed (anchor_on_sigil returns an error). The pipeline runs end-to-end without it, but availability is not proven until the Weft root is anchored on-chain. For production, wire anchor_on_sigil to a real Sigil RPC client.

How the primitives compose

  • Weft chunks, hashes, and signs the corpus. The root is a single 32-byte commitment.
  • The Strand named corpora-catalog is the discoverable index. Every published dataset gets a signed block recording its root and producer.
  • Sigil anchors the root, giving downstream auditors a proof that the dataset existed at a particular block height.
  • weave-dht is the byte-level transport for chunks and manifests.

Why this design

  • Idempotent producer. Re-running the producer with the same bytes produces the same root and the same chunk ids. Re-publishes are free.
  • Verifiable consumer. The consumer never trusts bytes from the network — each chunk's BLAKE3 id is checked, the manifest's Ed25519 signature is verified, and the chain commitment is consulted.
  • Independent availability. Many peers can mirror the chunks. Loss of any one peer does not lose the corpus.

What to do next

  • Replace InMemoryDhtKv with a real weave-dht adapter (see DHT Binding).
  • Add MARS asset minting before publishing so the corpus has a registry id (see Sigil Anchoring).
  • Stream-publish from a file too large to fit in memory using WeftDhtStore::publish_bytes over a chunked reader.
  • Use Semantic Search to build an index over a fetched corpus.

Semantic Search

Build a RAG-style document search using Basis, Strand, and Gnosis.

Production Release

Release gates for shipping Weave as production peer-to-peer infrastructure.

On this page

What you'll build
Prerequisites
Build the pipeline
How the primitives compose
Why this design
What to do next