feat(consensus): add RNG wire protocol and harvest logic

- Serialize full ExtendedPosition in share() and propose()
- Deserialize ExtendedPosition in PeerImp using fromSerialIter()
- Add harvestRngData() to collect commits/reveals from peer proposals
- Conditionally call harvest via if constexpr for test compatibility

src/ripple/app/consensus/RCLConsensus.cpp

@@ -36,6 +36,7 @@
 #include <ripple/app/misc/ValidatorKeys.h>
 #include <ripple/app/misc/ValidatorList.h>
 #include <ripple/basics/random.h>
+#include <ripple/crypto/csprng.h>
 #include <ripple/beast/core/LexicalCast.h>
 #include <ripple/consensus/LedgerTiming.h>
 #include <ripple/nodestore/DatabaseShard.h>
@@ -165,9 +166,12 @@ RCLConsensus::Adaptor::share(RCLCxPeerPos const& peerPos)
     prop.set_proposeseq(proposal.proposeSeq());
     prop.set_closetime(proposal.closeTime().time_since_epoch().count());
 
-    prop.set_currenttxhash(
-        proposal.position().txSetHash.begin(),
-        proposal.position().txSetHash.size());
+    // Serialize full ExtendedPosition (includes RNG leaves)
+    Serializer positionData;
+    proposal.position().add(positionData);
+    auto const posSlice = positionData.slice();
+    prop.set_currenttxhash(posSlice.data(), posSlice.size());
+
     prop.set_previousledger(
         proposal.prevLedger().begin(), proposal.prevLedger().size());
 
@@ -210,9 +214,12 @@ RCLConsensus::Adaptor::propose(RCLCxPeerPos::Proposal const& proposal)
 
     protocol::TMProposeSet prop;
 
-    prop.set_currenttxhash(
-        proposal.position().txSetHash.begin(),
-        proposal.position().txSetHash.size());
+    // Serialize full ExtendedPosition (includes RNG leaves)
+    Serializer positionData;
+    proposal.position().add(positionData);
+    auto const posSlice = positionData.slice();
+    prop.set_currenttxhash(posSlice.data(), posSlice.size());
+
     prop.set_previousledger(
         proposal.prevLedger().begin(), proposal.prevLedger().size());
     prop.set_proposeseq(proposal.proposeSeq());
@@ -1049,6 +1056,170 @@ RCLConsensus::Adaptor::updateOperatingMode(std::size_t const positions) const
         app_.getOPs().setMode(OperatingMode::CONNECTED);
 }
 
+//------------------------------------------------------------------------------
+// RNG Helper Methods
+
+std::size_t
+RCLConsensus::Adaptor::quorumThreshold() const
+{
+    auto [quorum, trustedKeys] = getQuorumKeys();
+    // Use 80% quorum for RNG commit/reveal
+    return (trustedKeys.size() * 80 + 99) / 100;
+}
+
+bool
+RCLConsensus::Adaptor::hasQuorumOfCommits() const
+{
+    return pendingCommits_.size() >= quorumThreshold();
+}
+
+bool
+RCLConsensus::Adaptor::hasMinimumReveals() const
+{
+    return pendingReveals_.size() >= quorumThreshold();
+}
+
+bool
+RCLConsensus::Adaptor::hasAnyReveals() const
+{
+    return !pendingReveals_.empty();
+}
+
+uint256
+RCLConsensus::Adaptor::buildCommitSet()
+{
+    // Sort commits deterministically by public key
+    std::vector<std::pair<PublicKey, uint256>> sorted;
+    for (auto const& [nodeId, commit] : pendingCommits_)
+    {
+        auto it = nodeIdToKey_.find(nodeId);
+        if (it != nodeIdToKey_.end())
+            sorted.emplace_back(it->second, commit);
+    }
+    std::sort(
+        sorted.begin(),
+        sorted.end(),
+        [](auto const& a, auto const& b) { return a.first < b.first; });
+
+    Serializer s;
+    for (auto const& [key, commit] : sorted)
+    {
+        s.addVL(key.slice());
+        s.addBitString(commit);
+    }
+    return sha512Half(s.slice());
+}
+
+uint256
+RCLConsensus::Adaptor::buildEntropySet()
+{
+    // Sort reveals deterministically by public key
+    std::vector<std::pair<PublicKey, uint256>> sorted;
+    for (auto const& [nodeId, reveal] : pendingReveals_)
+    {
+        auto it = nodeIdToKey_.find(nodeId);
+        if (it != nodeIdToKey_.end())
+            sorted.emplace_back(it->second, reveal);
+    }
+    std::sort(
+        sorted.begin(),
+        sorted.end(),
+        [](auto const& a, auto const& b) { return a.first < b.first; });
+
+    Serializer s;
+    for (auto const& [key, reveal] : sorted)
+    {
+        s.addVL(key.slice());
+        s.addBitString(reveal);
+    }
+    return sha512Half(s.slice());
+}
+
+void
+RCLConsensus::Adaptor::generateEntropySecret()
+{
+    // Generate cryptographically secure random entropy
+    crypto_prng()(myEntropySecret_.data(), myEntropySecret_.size());
+    entropyFailed_ = false;
+}
+
+uint256
+RCLConsensus::Adaptor::getEntropySecret() const
+{
+    return myEntropySecret_;
+}
+
+void
+RCLConsensus::Adaptor::setEntropyFailed()
+{
+    entropyFailed_ = true;
+}
+
+PublicKey const&
+RCLConsensus::Adaptor::validatorKey() const
+{
+    return validatorKeys_.publicKey;
+}
+
+void
+RCLConsensus::Adaptor::clearRngState()
+{
+    pendingCommits_.clear();
+    pendingReveals_.clear();
+    nodeIdToKey_.clear();
+    myEntropySecret_ = uint256{};
+    entropyFailed_ = false;
+}
+
+void
+RCLConsensus::Adaptor::harvestRngData(
+    NodeID const& nodeId,
+    PublicKey const& publicKey,
+    ExtendedPosition const& position)
+{
+    // Store nodeId -> publicKey mapping for deterministic ordering
+    nodeIdToKey_[nodeId] = publicKey;
+
+    // Harvest commitment if present
+    if (position.myCommitment)
+    {
+        auto [it, inserted] = pendingCommits_.emplace(nodeId, *position.myCommitment);
+        if (!inserted && it->second != *position.myCommitment)
+        {
+            // Commitment changed - this is suspicious but could be from a
+            // restarted validator. Log and update.
+            JLOG(j_.warn()) << "Validator " << nodeId
+                            << " changed commitment from " << it->second
+                            << " to " << *position.myCommitment;
+            it->second = *position.myCommitment;
+        }
+        else if (inserted)
+        {
+            JLOG(j_.trace()) << "Harvested commitment from " << nodeId
+                             << ": " << *position.myCommitment;
+        }
+    }
+
+    // Harvest reveal if present
+    if (position.myReveal)
+    {
+        auto [it, inserted] = pendingReveals_.emplace(nodeId, *position.myReveal);
+        if (!inserted && it->second != *position.myReveal)
+        {
+            // Reveal changed - this should never happen for honest validators
+            JLOG(j_.warn()) << "Validator " << nodeId
+                            << " changed reveal from " << it->second
+                            << " to " << *position.myReveal;
+            it->second = *position.myReveal;
+        }
+        else if (inserted)
+        {
+            JLOG(j_.trace()) << "Harvested reveal from " << nodeId
+                             << ": " << *position.myReveal;
+        }
+    }
+}
+
 void
 RCLConsensus::startRound(
     NetClock::time_point const& now,

src/ripple/app/consensus/RCLConsensus.h

@@ -87,6 +87,15 @@ class RCLConsensus
         RCLCensorshipDetector<TxID, LedgerIndex> censorshipDetector_;
         NegativeUNLVote nUnlVote_;
 
+        // --- RNG Pipelined Storage ---
+        hash_map<NodeID, uint256> pendingCommits_;
+        hash_map<NodeID, uint256> pendingReveals_;
+        hash_map<NodeID, PublicKey> nodeIdToKey_;
+
+        // Ephemeral entropy secret (in-memory only, crash = non-revealer)
+        uint256 myEntropySecret_;
+        bool entropyFailed_ = false;
+
     public:
         using Ledger_t = RCLCxLedger;
         using NodeID_t = NodeID;
@@ -179,6 +188,67 @@ class RCLConsensus
             return parms_;
         }
 
+        // --- RNG Helper Methods ---
+
+        /** Get the quorum threshold (80% of trusted validators) */
+        std::size_t
+        quorumThreshold() const;
+
+        /** Check if we have quorum of commits */
+        bool
+        hasQuorumOfCommits() const;
+
+        /** Check if we have minimum reveals for consensus */
+        bool
+        hasMinimumReveals() const;
+
+        /** Check if we have any reveals at all */
+        bool
+        hasAnyReveals() const;
+
+        /** Build deterministic hash of all collected commits */
+        uint256
+        buildCommitSet();
+
+        /** Build deterministic hash of all collected reveals */
+        uint256
+        buildEntropySet();
+
+        /** Generate new entropy secret for this round */
+        void
+        generateEntropySecret();
+
+        /** Get the current entropy secret */
+        uint256
+        getEntropySecret() const;
+
+        /** Mark entropy as failed for this round */
+        void
+        setEntropyFailed();
+
+        /** Get our validator public key */
+        PublicKey const&
+        validatorKey() const;
+
+        /** Clear RNG state for new round */
+        void
+        clearRngState();
+
+        /** Harvest RNG data from a peer proposal.
+
+            Extracts commits and reveals from the proposal's ExtendedPosition
+            and stores them in pending collections for later processing.
+
+            @param nodeId The node ID of the proposer
+            @param publicKey The public key of the proposer
+            @param position The proposal's ExtendedPosition
+        */
+        void
+        harvestRngData(
+            NodeID const& nodeId,
+            PublicKey const& publicKey,
+            ExtendedPosition const& position);
+
     private:
         //---------------------------------------------------------------------
         // The following members implement the generic Consensus requirements

src/ripple/app/consensus/RCLCxPeerPos.h

@@ -154,6 +154,36 @@ struct ExtendedPosition
             ret["entropy_set"] = to_string(*entropySetHash);
         return ret;
     }
+
+    /** Deserialize from wire format.
+        Handles both legacy 32-byte hash and new extended format.
+    */
+    static ExtendedPosition
+    fromSerialIter(SerialIter& sit, std::size_t totalSize)
+    {
+        ExtendedPosition pos;
+        pos.txSetHash = sit.get256();
+
+        // Legacy format: exactly 32 bytes
+        if (totalSize == 32)
+            return pos;
+
+        // Extended format: has flags + optional fields
+        if (sit.empty())
+            return pos;
+
+        std::uint8_t flags = sit.get8();
+        if (flags & 0x01)
+            pos.commitSetHash = sit.get256();
+        if (flags & 0x02)
+            pos.entropySetHash = sit.get256();
+        if (flags & 0x04)
+            pos.myCommitment = sit.get256();
+        if (flags & 0x08)
+            pos.myReveal = sit.get256();
+
+        return pos;
+    }
 };
 
 // For logging/debugging - returns txSetHash as string

src/ripple/consensus/Consensus.h

@@ -796,6 +796,18 @@ Consensus<Adaptor>::peerProposalInternal(
             currPeerPositions_.emplace(peerID, newPeerPos);
     }
 
+    // Harvest RNG data from proposal if adaptor supports it
+    if constexpr (requires(Adaptor& a, PeerPosition_t const& pp) {
+                      a.harvestRngData(
+                          pp.proposal().nodeID(),
+                          pp.publicKey(),
+                          pp.proposal().position());
+                  })
+    {
+        adaptor_.harvestRngData(
+            peerID, newPeerPos.publicKey(), newPeerProp.position());
+    }
+
     if (newPeerProp.isInitial())
     {
         // Record the close time estimate

src/ripple/overlay/impl/PeerImp.cpp

@@ -1935,7 +1935,8 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMProposeSet> const& m)
         return;
     }
 
-    if (!stringIsUint256Sized(set.currenttxhash()) ||
+    // Position data must be at least 32 bytes (txSetHash), previous ledger exactly 32
+    if (set.currenttxhash().size() < 32 ||
         !stringIsUint256Sized(set.previousledger()))
     {
         JLOG(p_journal_.warn()) << "Proposal: malformed";
@@ -1955,13 +1956,18 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMProposeSet> const& m)
     if (!isTrusted && app_.config().RELAY_UNTRUSTED_PROPOSALS == -1)
         return;
 
-    uint256 const proposeHash{set.currenttxhash()};
+    // Deserialize ExtendedPosition (handles both legacy 32-byte and extended formats)
+    auto const positionSlice = makeSlice(set.currenttxhash());
+    SerialIter sit(positionSlice);
+    ExtendedPosition const position =
+        ExtendedPosition::fromSerialIter(sit, positionSlice.size());
+
     uint256 const prevLedger{set.previousledger()};
 
     NetClock::time_point const closeTime{NetClock::duration{set.closetime()}};
 
     uint256 const suppression = proposalUniqueId(
-        ExtendedPosition{proposeHash},
+        position,
         prevLedger,
         set.proposeseq(),
         closeTime,
@@ -2008,7 +2014,7 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMProposeSet> const& m)
         RCLCxPeerPos::Proposal{
             prevLedger,
             set.proposeseq(),
-            ExtendedPosition{proposeHash},
+            position,
             closeTime,
             app_.timeKeeper().closeTime(),
             calcNodeID(app_.validatorManifests().getMasterKey(publicKey))});