feat: add AggregateCountOnRange query for provable count trees

docs/book/book.toml

@@ -12,4 +12,4 @@ command = "mdbook-mermaid"
 
 [output.html]
 additional-css = ["lang-selector.css"]
-additional-js = ["mermaid.min.js", "mermaid-init.js", "lang-selector.js"]
+additional-js = ["mermaid.min.js", "mermaid-fixup.js", "mermaid-init.js", "lang-selector.js"]

docs/book/mermaid-fixup.js

@@ -0,0 +1,26 @@
+// Client-side fallback: converts `<pre><code class="language-mermaid">...`
+// blocks (raw mdbook output when the mdbook-mermaid preprocessor isn't run)
+// into `<pre class="mermaid">...</pre>` blocks that mermaid.js will render.
+//
+// Safe to leave enabled even when the preprocessor IS run — preprocessor
+// output already uses `<pre class="mermaid">`, so the selector below finds
+// nothing and the script is a no-op.
+(() => {
+    function fixup() {
+        const blocks = document.querySelectorAll('pre > code.language-mermaid');
+        blocks.forEach((code) => {
+            const pre = code.parentElement;
+            const replacement = document.createElement('pre');
+            replacement.className = 'mermaid';
+            // textContent decodes HTML entities (&lt; → <, &amp; → &, etc.)
+            replacement.textContent = code.textContent;
+            pre.replaceWith(replacement);
+        });
+    }
+
+    if (document.readyState === 'loading') {
+        document.addEventListener('DOMContentLoaded', fixup);
+    } else {
+        fixup();
+    }
+})();

docs/book/src/SUMMARY.md

@@ -11,6 +11,7 @@
 - [The Proof System](proof-system.md)
 - [The Query System](query-system.md)
 - [Aggregate Sum Queries](aggregate-sum-queries.md)
+- [Aggregate Count Queries](aggregate-count-queries.md)
 - [Batch Operations](batch-operations.md)
 - [Cost Tracking](cost-tracking.md)
 - [The MMR Tree](mmr-tree.md)

docs/book/src/query-system.md

@@ -50,9 +50,15 @@ pub enum QueryItem {
     RangeAfter(RangeFrom<Vec<u8>>),         // (start..) exclusive start
     RangeAfterTo(Range<Vec<u8>>),           // (start..end) exclusive both
     RangeAfterToInclusive(RangeInclusive<Vec<u8>>), // (start..=end]
+    AggregateCountOnRange(Box<QueryItem>),         // Count-only — see Aggregate Count Queries
 }
 ```
 
+> **`AggregateCountOnRange`** is a terminal item: when present, it must be the **only**
+> item in the `Query`, and the query may not carry subqueries or pagination.
+> See [Aggregate Count Queries](aggregate-count-queries.md) for the full
+> contract — it is restricted to provable count trees.
+
 Example queries:
 
 Merk tree (sorted): `alice  bob  carol  dave  eve  frank`

grovedb-bulk-append-tree/src/proof/mod.rs

@@ -135,6 +135,13 @@ fn query_to_ranges(query: &Query, total_count: u64) -> Result<Vec<(u64, u64)>, B
                 }
                 (s, e)
             }
+            QueryItem::AggregateCountOnRange(_) => {
+                return Err(BulkAppendError::InvalidInput(
+                    "AggregateCountOnRange is only supported on provable count trees, \
+                     not on BulkAppendTree"
+                        .into(),
+                ));
+            }
         };
         ranges.push((start, end));
     }

grovedb-dense-fixed-sized-merkle-tree/src/proof/mod.rs

@@ -116,6 +116,13 @@ pub(crate) fn query_to_positions(query: &Query, count: u16) -> Result<Vec<u16>,
                     positions.insert(p);
                 }
             }
+            QueryItem::AggregateCountOnRange(_) => {
+                return Err(DenseMerkleError::InvalidProof(
+                    "AggregateCountOnRange is only supported on provable count trees, \
+                     not on dense fixed-size merkle trees"
+                        .into(),
+                ));
+            }
         }
     }
 

grovedb-query/Cargo.toml

@@ -26,6 +26,7 @@ grovedb-storage = { version = "4.0.0", path = "../storage", optional = true }
 
 [dev-dependencies]
 assert_matches = { workspace = true }
+serde_test = "1.0"
 
 [features]
 default = []

grovedb-query/src/proofs/encoding.rs

@@ -150,6 +150,13 @@ impl Encode for Op {
                 dest.write_all(value_hash)?;
                 count.encode_into(dest)?;
             }
+            Op::Push(Node::HashWithCount(kv_hash, left_child_hash, right_child_hash, count)) => {
+                dest.write_all(&[0x1e])?;
+                dest.write_all(kv_hash)?;
+                dest.write_all(left_child_hash)?;
+                dest.write_all(right_child_hash)?;
+                count.encode_into(dest)?;
+            }
             Op::Push(Node::KVValueHashFeatureTypeWithChildHash(
                 key,
                 value,
@@ -309,6 +316,18 @@ impl Encode for Op {
                 dest.write_all(value_hash)?;
                 count.encode_into(dest)?;
             }
+            Op::PushInverted(Node::HashWithCount(
+                kv_hash,
+                left_child_hash,
+                right_child_hash,
+                count,
+            )) => {
+                dest.write_all(&[0x1f])?;
+                dest.write_all(kv_hash)?;
+                dest.write_all(left_child_hash)?;
+                dest.write_all(right_child_hash)?;
+                count.encode_into(dest)?;
+            }
             Op::PushInverted(Node::KVValueHashFeatureTypeWithChildHash(
                 key,
                 value,
@@ -377,6 +396,9 @@ impl Encode for Op {
             Op::Push(Node::KVDigestCount(key, _, count)) => {
                 2 + key.len() + HASH_LENGTH + count.encoding_length()?
             }
+            Op::Push(Node::HashWithCount(_, _, _, count)) => {
+                1 + 3 * HASH_LENGTH + count.encoding_length()?
+            }
             Op::Push(Node::KVValueHashFeatureTypeWithChildHash(key, value, _, feature_type, _)) => {
                 let header = if value.len() < 65536 { 4 } else { 6 };
                 header
@@ -419,6 +441,9 @@ impl Encode for Op {
             Op::PushInverted(Node::KVDigestCount(key, _, count)) => {
                 2 + key.len() + HASH_LENGTH + count.encoding_length()?
             }
+            Op::PushInverted(Node::HashWithCount(_, _, _, count)) => {
+                1 + 3 * HASH_LENGTH + count.encoding_length()?
+            }
             Op::PushInverted(Node::KVValueHashFeatureTypeWithChildHash(
                 key,
                 value,
@@ -722,6 +747,38 @@ impl Decode for Op {
                     child_hash,
                 ))
             }
+            0x1e => {
+                let mut kv_hash = [0; HASH_LENGTH];
+                input.read_exact(&mut kv_hash)?;
+                let mut left_child_hash = [0; HASH_LENGTH];
+                input.read_exact(&mut left_child_hash)?;
+                let mut right_child_hash = [0; HASH_LENGTH];
+                input.read_exact(&mut right_child_hash)?;
+                let count: u64 = Decode::decode(&mut input)?;
+
+                Self::Push(Node::HashWithCount(
+                    kv_hash,
+                    left_child_hash,
+                    right_child_hash,
+                    count,
+                ))
+            }
+            0x1f => {
+                let mut kv_hash = [0; HASH_LENGTH];
+                input.read_exact(&mut kv_hash)?;
+                let mut left_child_hash = [0; HASH_LENGTH];
+                input.read_exact(&mut left_child_hash)?;
+                let mut right_child_hash = [0; HASH_LENGTH];
+                input.read_exact(&mut right_child_hash)?;
+                let count: u64 = Decode::decode(&mut input)?;
+
+                Self::PushInverted(Node::HashWithCount(
+                    kv_hash,
+                    left_child_hash,
+                    right_child_hash,
+                    count,
+                ))
+            }
             0x1d => {
                 let key_len: u8 = Decode::decode(&mut input)?;
                 let mut key = vec![0; key_len as usize];
@@ -2217,4 +2274,98 @@ mod test {
         let decoded = Op::decode(&bytes[..]).expect("decode failed");
         assert_eq!(decoded, op);
     }
+
+    #[test]
+    fn encode_decode_push_hash_with_count() {
+        // (kv_hash, left_child_hash, right_child_hash, count) — the
+        // self-verifying compressed-subtree variant for AggregateCountOnRange.
+        let op = Op::Push(Node::HashWithCount(
+            [0xAA; HASH_LENGTH],
+            [0xBB; HASH_LENGTH],
+            [0xCC; HASH_LENGTH],
+            42,
+        ));
+        // 1 opcode + 3 * 32 hashes + varint(42) = 1 + 96 + 1 = 98
+        let expected_length = 1 + 3 * HASH_LENGTH + ed::Encode::encoding_length(&42u64).unwrap();
+        assert_eq!(op.encoding_length(), expected_length);
+
+        let mut bytes = vec![];
+        op.encode_into(&mut bytes).unwrap();
+        assert_eq!(bytes.len(), expected_length);
+        assert_eq!(bytes[0], 0x1e); // Push HashWithCount opcode
+
+        let decoded = Op::decode(&bytes[..]).expect("decode failed");
+        assert_eq!(decoded, op);
+    }
+
+    #[test]
+    fn encode_decode_push_inverted_hash_with_count() {
+        let op = Op::PushInverted(Node::HashWithCount(
+            [0x11; HASH_LENGTH],
+            [0x22; HASH_LENGTH],
+            [0x33; HASH_LENGTH],
+            u64::MAX,
+        ));
+        let expected_length = 1 + 3 * HASH_LENGTH + ed::Encode::encoding_length(&u64::MAX).unwrap();
+        assert_eq!(op.encoding_length(), expected_length);
+
+        let mut bytes = vec![];
+        op.encode_into(&mut bytes).unwrap();
+        assert_eq!(bytes.len(), expected_length);
+        assert_eq!(bytes[0], 0x1f); // PushInverted HashWithCount opcode
+
+        let decoded = Op::decode(&bytes[..]).expect("decode failed");
+        assert_eq!(decoded, op);
+    }
+
+    #[test]
+    fn encode_decode_hash_with_count_zero_count_zero_children() {
+        // count = 0 (encodes to a 1-byte varint), all-zero hashes — represents
+        // a leaf-shaped collapsed subtree with no children.
+        let op = Op::Push(Node::HashWithCount(
+            [0u8; HASH_LENGTH],
+            [0u8; HASH_LENGTH],
+            [0u8; HASH_LENGTH],
+            0,
+        ));
+        let mut bytes = vec![];
+        op.encode_into(&mut bytes).unwrap();
+        assert_eq!(bytes[0], 0x1e);
+        let decoded = Op::decode(&bytes[..]).expect("decode failed");
+        assert_eq!(decoded, op);
+    }
+
+    #[test]
+    fn decoder_with_hash_with_count_mixed_with_other_count_nodes() {
+        // Round-trip a small Op stream containing HashWithCount alongside the
+        // existing count-bearing variants — exercises the Decoder iterator
+        // boundary handling for the new variants.
+        let ops = vec![
+            Op::Push(Node::HashWithCount(
+                [1; HASH_LENGTH],
+                [2; HASH_LENGTH],
+                [3; HASH_LENGTH],
+                7,
+            )),
+            Op::Push(Node::KVDigestCount(vec![0xAB], [4; HASH_LENGTH], 1)),
+            Op::Parent,
+            Op::Push(Node::Hash([5; HASH_LENGTH])),
+            Op::Child,
+            Op::PushInverted(Node::HashWithCount(
+                [6; HASH_LENGTH],
+                [7; HASH_LENGTH],
+                [8; HASH_LENGTH],
+                12345,
+            )),
+        ];
+
+        let mut encoded = vec![];
+        for op in &ops {
+            op.encode_into(&mut encoded).unwrap();
+        }
+
+        let decoder = Decoder::new(&encoded);
+        let decoded_ops: Result<Vec<Op>, _> = decoder.collect();
+        assert_eq!(decoded_ops.unwrap(), ops);
+    }
 }

grovedb-query/src/proofs/mod.rs

@@ -127,6 +127,30 @@ pub enum Node {
     ///
     /// Contains: `(key, value, value_hash, feature_type, child_hash)`
     KVValueHashFeatureTypeWithChildHash(Vec<u8>, Vec<u8>, CryptoHash, TreeFeatureType, CryptoHash),
+
+    /// A self-verifying compressed subtree for `AggregateCountOnRange` proofs
+    /// against a `ProvableCountTree` / `ProvableCountSumTree`.
+    ///
+    /// Encodes the subtree's *root* node as `(kv_hash, left_child_hash,
+    /// right_child_hash, count)`. The verifier reconstructs the subtree's
+    /// root `node_hash` as
+    /// `node_hash_with_count(kv_hash, left_child_hash, right_child_hash, count)`
+    /// and uses that hash exactly as `Hash(...)` would. Because `count` is
+    /// part of that recomputation, a forged count produces a different hash
+    /// and the parent's Merkle-root check fails — the count is therefore
+    /// cryptographically committed by the parent's hash chain, not just
+    /// trusted on faith.
+    ///
+    /// Used to collapse an entire fully-inside subtree into a single proof
+    /// node: the verifier doesn't need any per-key information (the parent
+    /// boundary nodes already established that every key under here is
+    /// in-range), so we hand it the four hashes plus the count.
+    ///
+    /// `left_child_hash` / `right_child_hash` are the all-zero `NULL_HASH`
+    /// when the subtree's root has no left / right child respectively.
+    ///
+    /// Contains: `(kv_hash, left_child_hash, right_child_hash, count)`
+    HashWithCount(CryptoHash, CryptoHash, CryptoHash, u64),
 }
 
 use std::fmt;
@@ -185,6 +209,13 @@ impl fmt::Display for Node {
                 hex::encode(value_hash),
                 count
             ),
+            Node::HashWithCount(kv_hash, left_child_hash, right_child_hash, count) => format!(
+                "HashWithCount(kv_hash=HASH[{}], left=HASH[{}], right=HASH[{}], count={})",
+                hex::encode(kv_hash),
+                hex::encode(left_child_hash),
+                hex::encode(right_child_hash),
+                count
+            ),
             Node::KVValueHashFeatureTypeWithChildHash(
                 key,
                 value,