open-standards/joule-code/examples/transport-moqt/tests/transport.rs

//! CommunicationOS (MoQT) network transport for a Joule Code collaboration
//! (SPEC §7, spec/collab.md — the last federation layer).
//!
//! The full federation stack, end to end, in-process:
//!
//!   BFT consensus (`LockstepCluster`)  →  committed, op_id-ordered coding ops
//!        →  published as MoQT objects over CommunicationOS's transport (`inv-moq`)
//!        →  each object carries a real Ed25519/BLAKE3 *signed receipt*
//!        →  a subscriber receives every op, verifies the signed receipts against
//!           the publisher's key, and recovers the original `SignedOp` intact.
//!
//! The transport is `inv-moq` — joulesperbit's MoQT (Media over QUIC Transport)
//! wire protocol + JWP receipt extension, the CommunicationOS broadcast carrier.
//! We drive it through its `MockFabric`, which runs the real publish/subscribe/
//! receipt machinery in-process (real signatures, real BLAKE3 merkle receipts) —
//! no QUIC, no network. The `quic` feature of `inv-moq` is the real-wire path;
//! the trait surface (`Publisher` / `Subscriber` / `GroupHandle`) is identical.

use bytes::Bytes;
use ed25519_dalek::SigningKey;
use std::sync::Arc;

use joule_moq::endpoint::{Publisher, Subscriber};
use joule_moq::mock::{MockFabric, MockPublisher, MockSubscriber};
use joule_moq::receipt::ReceiptVerifier;
use joule_moq::track::{GroupId, TrackName, TrackNamespace};

use joule_code_cargo::{Anchor, CodingOp, Edit, SignedOp};
use joule_code_collab::{LockstepCluster, Membership};
use joule_code_schema::{ContentId, Tier, ALG_ED25519};

/// Build and sign a coding op against a prior state.
fn signed_op(sk: &SigningKey, prior: &str, path: &str, intent: &str) -> (SignedOp, ContentId) {
    let prior = ContentId(prior.to_string());
    let op = CodingOp {
        alg: ALG_ED25519,
        prior: prior.clone(),
        edits: vec![Edit {
            path: path.to_string(),
            anchor: Anchor::ByteRange { start: 0, end: 0 },
            replacement: intent.as_bytes().to_vec(),
        }],
        author: sk.verifying_key().to_bytes(),
        intent: intent.to_string(),
        tier: Tier::L4,
    };
    let signed = op.sign(sk).expect("sign op");
    (signed, prior)
}

#[tokio::test]
async fn committed_ops_replicate_over_communicationos_moqt_with_signed_receipts() {
    // ── 1. Consensus ──────────────────────────────────────────────────────────
    // A single-member BFT cluster commits two coding ops. The committed log is
    // op_id-ordered and identical across every honest node; here we take that log
    // as the payload stream to replicate to collaborators.
    let author = SigningKey::from_bytes(&[3u8; 32]);
    let membership = Membership::new(vec![author.verifying_key().to_bytes()]);
    let mut cluster = LockstepCluster::new(membership, 4, []);

    let (op_a, prior_a) = signed_op(&author, "b3:genesis", "src/a.rs", "fn a() {}");
    let (op_b, prior_b) = signed_op(&author, "b3:genesis", "src/b.rs", "fn b() {}");
    cluster.submit_round(vec![(op_a, prior_a), (op_b, prior_b)]);

    let committed: Vec<SignedOp> = cluster.agreed_log().iter().map(|c| c.op.clone()).collect();
    assert_eq!(committed.len(), 2, "both ops commit under honest supermajority");

    // ── 2. Transport: publish committed ops over CommunicationOS MoQT ─────────
    let fabric = Arc::new(MockFabric::new());
    // The publisher (this cluster endpoint) signs every object + the group root.
    let endpoint_key = SigningKey::from_bytes(&[7u8; 32]);
    let endpoint_vk = endpoint_key.verifying_key();
    let mut publisher = MockPublisher::new(fabric.clone(), "joule-code/cluster-0", endpoint_key);
    let mut subscriber = MockSubscriber::new(fabric.clone());

    // Namespace/track for this collaboration's committed coding-op stream.
    let ns = TrackNamespace::from_strs(["joule-code", "collab"]);
    let track = TrackName::from_str("ops");

    // A collaborator subscribes, requiring receipts (rejects an unsigned source).
    let mut sub = subscriber
        .subscribe(ns.clone(), track.clone(), true)
        .await
        .expect("subscribe");

    publisher.announce(ns.clone()).await.expect("announce");
    let mut group = publisher
        .open_group(&ns, &track, GroupId(0))
        .await
        .expect("open group");
    for op in &committed {
        // Each committed op becomes one MoQT object. The `joules_micro` field is
        // the transport's egress-energy meter for the object (compute energy
        // lives in the op's own receipt; this is the wire cost).
        let payload = Bytes::from(serde_json::to_vec(op).expect("encode op"));
        group
            .push_object(payload, 1_000)
            .await
            .expect("push object");
    }
    let group_receipt = group.finalize().await.expect("finalize group");

    // ── 3. Receive + verify + recover ─────────────────────────────────────────
    tokio::task::yield_now().await;
    let mut received = Vec::new();
    while let Ok(obj) = sub.rx.try_recv() {
        received.push(obj);
    }
    assert_eq!(
        received.len(),
        committed.len(),
        "every committed op is delivered over the transport"
    );

    // Verify the signed receipts against the publisher's key, then recover the op.
    let verifier = ReceiptVerifier::with_anchors(vec![endpoint_vk]);
    let root = verifier
        .verify_group(&group_receipt)
        .expect("group receipt verifies under the publisher key");
    for (obj, original) in received.iter().zip(committed.iter()) {
        let receipt = obj.receipt.as_ref().expect("object receipt present");
        verifier
            .verify_payload(&obj.payload, receipt, &root)
            .expect("object receipt verifies against the group root");
        let recovered: SignedOp =
            serde_json::from_slice(&obj.payload).expect("recover SignedOp from payload");
        assert_eq!(
            &recovered, original,
            "the coding op round-trips intact over the CommunicationOS transport"
        );
    }
}

/// A receipt-requiring subscriber must reject objects whose signed group receipt
/// does not chain to a trusted publisher key — ownerless verification, no central
/// authority deciding who is authentic.
#[tokio::test]
async fn untrusted_publisher_is_rejected() {
    let fabric = Arc::new(MockFabric::new());
    let attacker = SigningKey::from_bytes(&[9u8; 32]);
    let trusted = SigningKey::from_bytes(&[1u8; 32]);

    let mut publisher = MockPublisher::new(fabric.clone(), "imposter", attacker);
    let mut subscriber = MockSubscriber::new(fabric.clone());
    let ns = TrackNamespace::from_strs(["joule-code", "collab"]);
    let track = TrackName::from_str("ops");
    let _sub = subscriber
        .subscribe(ns.clone(), track.clone(), true)
        .await
        .expect("subscribe");

    let mut group = publisher
        .open_group(&ns, &track, GroupId(0))
        .await
        .expect("open group");
    group
        .push_object(Bytes::from_static(b"forged"), 1)
        .await
        .expect("push");
    let receipt = group.finalize().await.expect("finalize");

    // The collaborator only trusts a different key; the forged group fails.
    let verifier = ReceiptVerifier::with_anchors(vec![trusted.verifying_key()]);
    assert!(
        verifier.verify_group(&receipt).is_err(),
        "an untrusted publisher's group receipt must not verify"
    );
}