//! 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 = 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" ); }