fix(agent-mesh): make MerkleAuditChain root reproducible by an independent rebuild

BREAKING_CHANGES.md

@@ -5,6 +5,41 @@ entries appear first.
 
 ---
 
+## MerkleAuditChain now records a canonical, reproducible Merkle root
+
+**Date:** TBD (next release of `microsoft/agent-governance-toolkit`)
+
+**Affected:**
+
+- `agent-governance-python` (`agentmesh.governance.audit`):
+  `MerkleAuditChain.add_entry`, `MerkleAuditChain.get_root_hash`,
+  `AuditLog.export` (`merkle_root` / `chain_root` fields), and Merkle inclusion
+  proofs from `MerkleAuditChain.get_proof`.
+
+**What changed:**
+
+The incremental tree builder padded interior levels with the leaf sentinel
+`'0' * 64` rather than the empty-subtree constant `E(k)` (where `E(0) = '0' * 64`
+and `E(k+1) = SHA-256(E(k) || E(k))`). The recorded root therefore diverged from
+a from-scratch rebuild of the same leaves for most chain sizes (22 of the first
+32: n = 5, 6, 9-14, 17-30, and so on), so an exported `merkle_root` could not be
+reproduced by a verifier that rebuilt the tree. Interior padding now uses `E(k)`,
+so the incremental root, a from-scratch rebuild, and an independent textbook
+recomputation all agree, and the update stays amortized O(log n) per append
+(worst-case O(n) when tree capacity doubles).
+
+**Impact / required action:**
+
+For the affected chain sizes the exported `merkle_root` (and the corresponding
+inclusion proofs) now differ from values produced by earlier releases. Consumers
+who archived `AuditLog.export()` output as compliance evidence will see a
+mismatch when re-verifying old exports against a current build. Re-export and
+re-anchor any archived roots, or pin the prior version to verify pre-existing
+evidence. Entry hashes and the linear `previous_hash` chain (`verify_chain`) are
+unchanged.
+
+---
+
 ## acs-generator is now a CLI-only package
 
 **Date:** TBD (next `acs-generator` release)

agent-governance-python/agent-mesh/CHANGELOG.md

@@ -5,6 +5,20 @@ All notable changes to AgentMesh will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+
+- `MerkleAuditChain` now records a canonical, reproducible Merkle root. The
+  incremental `add_entry` padded interior tree levels with the leaf sentinel
+  `'0' * 64` instead of the empty-subtree constant `E(k)`, so the recorded root
+  diverged from a from-scratch rebuild for most chain sizes (22 of the first 32)
+  and an exported `merkle_root` could not be reproduced by an independent
+  verifier. Interior padding now uses `E(k)`, keeping the update amortized
+  O(log n) per append (worst-case O(n) when tree capacity doubles). Exported
+  `merkle_root` values change for the affected sizes; see
+  `BREAKING_CHANGES.md`.
+
 ## [1.0.0-alpha.1] - 2026-02-01
 
 ### Added

agent-governance-python/agent-mesh/src/agentmesh/governance/audit.py

@@ -279,9 +279,38 @@ def __init__(self):
         self._entries: list[AuditEntry] = []
         self._tree: list[list[MerkleNode]] = []
         self._root_hash: Optional[str] = None
+        # E(k): the Merkle hash of a fully zero-padded subtree of height k.
+        # E(0) is the zero-leaf sentinel; E(k+1) = SHA-256(E(k) || E(k)).
+        # Grown lazily by _empty_subtree_hash().
+        self._empty_hashes: list[str] = ['0' * 64]
+
+    def _empty_subtree_hash(self, level: int) -> str:
+        """Return E(level): the canonical hash of an all-zero subtree of that height.
+
+        Interior padding must use E(level), not the leaf sentinel E(0). Padding an
+        interior level with a bare '0'*64 makes the incremental root diverge from a
+        from-scratch rebuild (which pads at the leaf level and hashes the zeros
+        upward), so an exported ``merkle_root`` could not be reproduced by a verifier.
+        """
+        while len(self._empty_hashes) <= level:
+            prev = self._empty_hashes[-1]
+            self._empty_hashes.append(hashlib.sha256((prev + prev).encode()).hexdigest())
+        return self._empty_hashes[level]
 
     def add_entry(self, entry: AuditEntry) -> None:
-        """Add an entry and update the Merkle tree incrementally."""
+        """Add an entry and update the Merkle tree incrementally.
+
+        Interior padding uses the empty-subtree constant :meth:`_empty_subtree_hash`
+        (E(k)) rather than a bare ``'0' * 64`` sentinel at every level. The former
+        code padded interior levels with E(0), so the incremental root diverged from
+        a from-scratch rebuild (which pads at the leaf level and hashes the zeros
+        upward) as soon as a real leaf's authentication path reached an interior
+        padding node (first observable at five entries). With E(k) padding the
+        incremental construction converges on the same canonical root as
+        :meth:`_rebuild_tree`, so an exported ``merkle_root`` is reproducible by a
+        verifier, while the update stays amortized O(log n) per append
+        (worst-case O(n) when tree capacity doubles).
+        """
         # Set previous hash
         if self._entries:
             entry.previous_hash = self._entries[-1].entry_hash
@@ -308,21 +337,24 @@ def add_entry(self, entry: AuditEntry) -> None:
         # Check if we need to expand the tree capacity
         capacity = len(self._tree[0])
         if n > capacity:
-            # Double capacity: pad leaves with empty nodes, add new tree level
+            # Double capacity: pad each level with its own empty-subtree
+            # constant E(level), then add a new root level.
             for level_idx in range(len(self._tree)):
                 self._tree[level_idx].extend(
-                    [MerkleNode(hash="0" * 64) for _ in range(len(self._tree[level_idx]))]
+                    [MerkleNode(hash=self._empty_subtree_hash(level_idx))
+                     for _ in range(len(self._tree[level_idx]))]
                 )
-            # Add new root level
+            # Add new root level. The old root's sibling is an all-zero subtree
+            # at the old top level; the new level's spare slot is one higher.
             old_root = self._tree[-1][0]
-            empty_node = MerkleNode(hash="0" * 64)
-            combined = old_root.hash + empty_node.hash
+            empty_sibling = MerkleNode(hash=self._empty_subtree_hash(len(self._tree) - 1))
+            combined = old_root.hash + empty_sibling.hash
             new_root = MerkleNode(
                 hash=hashlib.sha256(combined.encode()).hexdigest(),
                 left_child=old_root.hash,
-                right_child=empty_node.hash,
+                right_child=empty_sibling.hash,
             )
-            self._tree.append([new_root, MerkleNode(hash="0" * 64)])
+            self._tree.append([new_root, MerkleNode(hash=self._empty_subtree_hash(len(self._tree)))])
 
         # Place new leaf
         leaf_idx = n - 1
@@ -372,13 +404,15 @@ def _rebuild_tree(self) -> None:
 
         self._tree = [leaves]
 
-        # Build tree bottom-up
+        # Build tree bottom-up. Leaves are padded to a power of two above, so
+        # every level has an even length and each node always has a right
+        # sibling; no odd-duplication fallback is needed.
         current_level = leaves
         while len(current_level) > 1:
             next_level = []
             for i in range(0, len(current_level), 2):
                 left = current_level[i]
-                right = current_level[i + 1] if i + 1 < len(current_level) else left
+                right = current_level[i + 1]
 
                 combined = left.hash + right.hash
                 parent_hash = hashlib.sha256(combined.encode()).hexdigest()

agent-governance-python/agent-mesh/tests/test_merkle_audit_root_reproducible.py

@@ -0,0 +1,124 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+"""Regression tests for ``MerkleAuditChain`` root reproducibility.
+
+The root recorded incrementally on each :meth:`MerkleAuditChain.add_entry`
+must equal a from-scratch rebuild over the same entries, so that an exported
+``merkle_root`` is reproducible by an independent verifier.
+
+Before the fix, ``add_entry`` padded *interior* tree levels with singleton
+zero nodes while ``_rebuild_tree`` padded at the *leaf* level and hashed the
+padding upward. The two constructions produced different roots once a real
+leaf's authentication path reached an interior padding node, first observable
+at five entries (and again at 6, 9, ...).
+"""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+
+from agentmesh.governance.audit import AuditEntry, MerkleAuditChain
+
+
+def _entry(i: int) -> AuditEntry:
+    return AuditEntry(
+        event_type="tool_invocation",
+        agent_did=f"did:mesh:agent-{i}",
+        action=f"act-{i}",
+        resource=f"res-{i}",
+    )
+
+
+def _textbook_merkle_root(leaf_hashes: list[str]) -> str | None:
+    """Independent, pure-hashlib textbook Merkle root (zero-leaf padded to a
+    power of two).
+
+    Deliberately shares no code with :class:`MerkleAuditChain`, so the
+    reproducibility assertions cannot pass by comparing the implementation with
+    itself.
+    """
+    if not leaf_hashes:
+        return None
+    level = list(leaf_hashes)
+    while len(level) & (len(level) - 1) != 0:
+        level.append('0' * 64)
+    while len(level) > 1:
+        level = [
+            hashlib.sha256((level[i] + level[i + 1]).encode()).hexdigest()
+            for i in range(0, len(level), 2)
+        ]
+    return level[0]
+
+
+class TestMerkleRootReproducible:
+    def test_incremental_root_matches_independent_recompute(self):
+        # The incremental root must equal a from-scratch textbook recomputation
+        # over the same leaves, across several capacity doublings and every size
+        # that diverged before the fix (5, 6, 9-14, 17-30).
+        for n in range(1, 33):
+            chain = MerkleAuditChain()
+            for i in range(n):
+                chain.add_entry(_entry(i))
+            incremental = chain.get_root_hash()
+            independent = _textbook_merkle_root([e.entry_hash for e in chain._entries])
+            assert incremental == independent, (
+                f"n={n}: incremental root {incremental} != independent recompute {independent}"
+            )
+
+    def test_incremental_root_matches_full_rebuild(self):
+        # The incremental construction and the from-scratch _rebuild_tree are
+        # separate code paths that must converge on the same canonical root.
+        # This deliberately calls the private _rebuild_tree: the divergence being
+        # regression-tested is *between* the two internal constructions, so the
+        # black-box public API alone cannot exercise it (the independent textbook
+        # recompute in the sibling test covers the public-API reproducibility).
+        for n in (5, 6, 9, 13, 16, 20):
+            chain = MerkleAuditChain()
+            for i in range(n):
+                chain.add_entry(_entry(i))
+            incremental = chain.get_root_hash()
+            chain._rebuild_tree()
+            assert chain.get_root_hash() == incremental, f"n={n}: rebuild disagrees with incremental"
+
+    def test_empty_chain_root_is_none(self):
+        chain = MerkleAuditChain()
+        assert chain.get_root_hash() is None
+        assert _textbook_merkle_root([]) is None
+
+    def test_five_entries_is_the_minimal_reproducer(self):
+        chain = MerkleAuditChain()
+        for i in range(5):
+            chain.add_entry(_entry(i))
+        recorded = chain.get_root_hash()
+        # Feeding the same entries to a fresh chain must yield the same root.
+        # Deep-copy each entry: add_entry rewrites previous_hash/entry_hash in
+        # place, so passing the originals would mutate chain._entries and make
+        # the independent recompute below no longer independent.
+        fresh = MerkleAuditChain()
+        for entry in list(chain._entries):
+            fresh.add_entry(copy.deepcopy(entry))
+        assert fresh.get_root_hash() == recorded
+        # And an independent recompute over the recorded entries must agree.
+        assert _textbook_merkle_root([e.entry_hash for e in chain._entries]) == recorded
+
+    def test_inclusion_proofs_verify_against_recorded_root(self):
+        chain = MerkleAuditChain()
+        entries = [_entry(i) for i in range(5)]
+        for entry in entries:
+            chain.add_entry(entry)
+        root = chain.get_root_hash()
+        for entry in entries:
+            proof = chain.get_proof(entry.entry_id)
+            assert proof is not None
+            assert chain.verify_proof(entry.entry_hash, proof, root)
+
+    def test_tampered_leaf_changes_the_rebuilt_root(self):
+        chain = MerkleAuditChain()
+        for i in range(5):
+            chain.add_entry(_entry(i))
+        recorded = chain.get_root_hash()
+        # Corrupt a stored leaf hash; a verifier that recomputes the root must
+        # observe a different value, i.e. tampering remains detectable.
+        chain._entries[1].entry_hash = hashlib.sha256(b"tampered").hexdigest()
+        assert _textbook_merkle_root([e.entry_hash for e in chain._entries]) != recorded