Branching and object registry rework (#11)

9 files changed, 1205 insertions, 223 deletions

diff --git a/src/CursorManager.cpp b/src/CursorManager.cpp
index 219ba2e..d96909b 100644
--- a/src/CursorManager.cpp
+++ b/src/CursorManager.cpp
@@ -5,25 +5,49 @@
 
 namespace nt
 {
-    CursorManager::CursorManager(IStorageBackend& backend)
+    CursorManager::CursorManager(IStorageBackend& backend,
+                                 LifecycleManager* lifecycles,
+                                 ObjectManager* objects)
         : backend_(backend)
+        , lifecycles_(lifecycles)
+        , objects_(objects)
     {}
 
+    // Finds the /system/snapshots/<H> entry that owns a handle's relation, or
+    // nullptr when the relation lives outside the snapshot namespace (tests use
+    // synthetic paths like {"relations", "villager"}).
+    static ObjectManager::registry* find_parent_snapshot(
+        ObjectManager& objects,
+        HandlerManager::handle* handle)
+    {
+        if (handle == nullptr || handle->object == nullptr
+            || handle->object->head == nullptr)
+            return nullptr;
+
+        const auto& p = handle->object->head->path;
+        if (p.size() != 5) return nullptr;
+        if (p[0] != "system" || p[1] != "snapshots" || p[3] != "relations")
+            return nullptr;
+
+        return objects.Find({ p[0], p[1], p[2] });
+    }
+
     void CursorManager::LoadPage(cursor* c)
     {
-        auto hashes = Merkle::Page(backend_, c->merkle_root, c->fetch_offset, PAGE_SIZE);
+        auto entries = Merkle<Hash32>::Page(backend_, c->merkle_root,
+                                             c->fetch_offset, PAGE_SIZE);
         c->page.clear();
         c->page_position = 0;
 
-        for (const auto& hash : hashes)
+        for (const auto& entry : entries)
         {
-            auto bytes = backend_.Get(hash);
+            auto bytes = backend_.Get(bin_to_hex(entry.payload));
             if (!bytes) continue;
             c->page.emplace_back(TupleCodec::Deserialize(*bytes));
         }
 
-        c->fetch_offset += hashes.size();
-        if (hashes.size() < PAGE_SIZE)
+        c->fetch_offset += entries.size();
+        if (entries.size() < PAGE_SIZE)
             c->exhausted = true;
     }
 
@@ -35,13 +59,28 @@ namespace nt
 
         const auto label = handle->object->head->type->label;
 
+        auto pin_snapshot = [&](cursor* c) {
+            if (lifecycles_ == nullptr || objects_ == nullptr) return;
+            auto* snap = find_parent_snapshot(*objects_, handle);
+            if (snap == nullptr) return;
+            lifecycles_->Pin(snap);
+            c->pinned_snapshot = snap;
+        };
+
         if (label == EPHEMERAL_RELATION)
         {
             // Cursor is created exhausted — the JOIN operator writes args and
-            // resets the state before the first probe.
+            // resets the state before the first probe. The generator is copied
+            // off the registry-owned ephemeral_object_type so Next() never has
+            // to dereference the handle again.
+            auto* etype = static_cast<ObjectManager::ephemeral_object_type*>(
+                handle->object->head->type.get());
+
             auto* c = new cursor();
-            c->handle = handle;
-            c->exhausted = true;
+            c->is_ephemeral = true;
+            c->generator    = etype->generator;
+            c->exhausted    = true;
+            pin_snapshot(c);
             return c;
         }
 
@@ -49,8 +88,8 @@ namespace nt
             return nullptr;
 
         auto* c = new cursor();
-        c->handle      = handle;
         c->merkle_root = merkle_root;
+        pin_snapshot(c);
 
         if (merkle_root.empty())
         {
@@ -72,11 +111,9 @@ namespace nt
 
         if (c->exhausted) return nullptr;
 
-        if (c->handle->object->head->type->label == EPHEMERAL_RELATION)
+        if (c->is_ephemeral)
         {
-            auto* etype = static_cast<ObjectManager::ephemeral_object_type*>(
-                c->handle->object->head->type.get());
-            c->page = etype->generator(c->args, c->fetch_offset, PAGE_SIZE);
+            c->page = c->generator(c->args, c->fetch_offset, PAGE_SIZE);
             c->page_position = 0;
             c->fetch_offset += c->page.size();
             if (c->page.size() < PAGE_SIZE)
@@ -92,6 +129,9 @@ namespace nt
 
     void CursorManager::Close(cursor* c)
     {
+        if (c == nullptr) { return; }
+        if (lifecycles_ != nullptr && c->pinned_snapshot != nullptr)
+            lifecycles_->Unpin(c->pinned_snapshot);
         delete c;
     }
 }
diff --git a/src/HandlerManager.cpp b/src/HandlerManager.cpp
index 8149ff2..cb3b723 100644
--- a/src/HandlerManager.cpp
+++ b/src/HandlerManager.cpp
@@ -2,6 +2,7 @@
 
 #include "IdentityManager.h"
 #include "LifecycleManager.h"
+#include "NamespaceReferenceManager.h"
 #include "PermissionsManager.h"
 
 namespace nt
@@ -14,16 +15,19 @@ namespace nt
     HandlerManager::HandlerManager(ObjectManager& objects,
                                        PermissionsManager& permissions,
                                        IdentityManager& identities,
-                                       LifecycleManager& lifecycles)
+                                       LifecycleManager& lifecycles,
+                                       NamespaceReferenceManager& references)
         : objects_(objects)
         , permissions_(permissions)
         , identities_(identities)
         , lifecycles_(lifecycles)
+        , references_(references)
     {}
 
     struct HandlerManager::handle* HandlerManager::Open(std::vector<std::string> object_path, void* connection_context)
     {
-        ObjectManager::registry* retrieved_object = objects_.Find((const std::vector<std::string>)object_path);
+        const auto resolved = references_.Resolve(std::move(object_path));
+        ObjectManager::registry* retrieved_object = objects_.Find(resolved);
         if (permissions_.Access(retrieved_object, (const void*)connection_context)
             && identities_.CanOpen(retrieved_object)
             && lifecycles_.Contention(retrieved_object)) {
diff --git a/src/LifecycleManager.cpp b/src/LifecycleManager.cpp
index a83c9e1..78e112d 100644
--- a/src/LifecycleManager.cpp
+++ b/src/LifecycleManager.cpp
@@ -1,21 +1,136 @@
 #include "LifecycleManager.h"
 
+#include "Merkle.h"
+#include "Types.h"
+
+#include <vector>
+
 namespace nt
 {
+    LifecycleManager::LifecycleManager(ObjectManager& objects,
+                                        IStorageBackend& storage)
+        : objects_(objects), storage_(storage)
+    {}
+
     void LifecycleManager::Monitor(ObjectManager::registry* object)
     {
         if (object != nullptr)
             ++object->head->handle_count;
     }
 
+    void LifecycleManager::Unmonitor(ObjectManager::registry* object)
+    {
+        if (object == nullptr || object->head->handle_count == 0) return;
+        --object->head->handle_count;
+        TryCollect(object);
+    }
+
+    void LifecycleManager::Pin(ObjectManager::registry* object)
+    {
+        if (object != nullptr)
+            ++object->head->reference_count;
+    }
+
+    void LifecycleManager::Unpin(ObjectManager::registry* object)
+    {
+        if (object == nullptr || object->head->reference_count == 0) return;
+        --object->head->reference_count;
+        TryCollect(object);
+    }
+
+    bool LifecycleManager::IsEligibleForGC(ObjectManager::registry* object) const
+    {
+        if (object == nullptr || object->head == nullptr) return false;
+        return object->head->handle_count == 0
+            && object->head->reference_count == 0;
+    }
+
     const bool LifecycleManager::Contention(ObjectManager::registry* object)
     {
         return object != nullptr;
     }
 
-    void LifecycleManager::Unmonitor(ObjectManager::registry* object)
+    void LifecycleManager::TryCollect(ObjectManager::registry* object)
     {
-        if (object != nullptr && object->head->handle_count > 0)
-            --object->head->handle_count;
+        if (!IsEligibleForGC(object)) return;
+        if (object->head == nullptr || object->head->type == nullptr) return;
+
+        // Named-lifetime types are managed explicitly and must not auto-collect:
+        //   BRANCH  — named branches persist until a future explicit-delete API
+        //             tears them down. Their counters going to zero just means
+        //             nobody currently holds them open.
+        //   SESSION — rnt_session_close is the sole removal path.
+        // Everything else (MULTIGROUP, BRANCH_TREE, RELATION,
+        // EPHEMERAL_RELATION, TRANSACTION) is purely structurally referenced
+        // and is GC-eligible by counters alone.
+        const auto label = object->head->type->label;
+        if (label == BRANCH || label == SESSION) return;
+
+        if (label == MULTIGROUP)   CascadeMultigroup(object);
+        if (label == BRANCH_TREE)  CascadeBranchTree(object);
+
+        objects_.Unregister(object->head->path);
+    }
+
+    void LifecycleManager::CascadeMultigroup(ObjectManager::registry* multigroup)
+    {
+        if (multigroup == nullptr || multigroup->head == nullptr) return;
+        const auto& mg_path = multigroup->head->path;
+        if (mg_path.size() < 3) return;
+
+        // /system/snapshots/<hash>/relations/<n> entries are the relation
+        // children pinned by this multigroup. Collect first, mutate after —
+        // Unpin can Unregister the entry we're standing on if its counters
+        // were already zero, and iteration must outlive that mutation.
+        std::vector<ObjectManager::registry*> children;
+        for (auto* cur = objects_.entries.get(); cur != nullptr; cur = cur->next.get())
+        {
+            const auto& p = cur->head->path;
+            if (p.size() != mg_path.size() + 2) continue;
+
+            bool prefix_match = true;
+            for (size_t i = 0; i < mg_path.size(); ++i)
+                if (p[i] != mg_path[i]) { prefix_match = false; break; }
+            if (!prefix_match) continue;
+            if (p[mg_path.size()] != "relations") continue;
+
+            children.push_back(cur);
+        }
+
+        for (auto* child : children) Unpin(child);
+    }
+
+    void LifecycleManager::CascadeBranchTree(ObjectManager::registry* branch_tree)
+    {
+        if (branch_tree == nullptr || branch_tree->head == nullptr) return;
+
+        auto* bt = dynamic_cast<ObjectManager::BranchTree*>(branch_tree->object.get());
+        if (bt == nullptr || bt->merkle_root.empty()) return;
+
+        // Page through the (mg_name, mg_hash) tree and collect each referenced
+        // multigroup before unpinning. Decoupling iteration from mutation
+        // keeps us safe against Unpin cascading further GC under our feet.
+        std::vector<ObjectManager::registry*> mgs;
+        constexpr size_t kPageSize = 1024;
+        size_t offset = 0;
+        while (true)
+        {
+            auto page = Merkle<std::string>::Page(storage_, bt->merkle_root,
+                                                   offset, kPageSize);
+            if (page.empty()) break;
+            for (const auto& entry : page)
+            {
+                static const Hash32 zero{};
+                if (entry.payload == zero) continue;
+                const std::string mg_hash = bin_to_hex(entry.payload);
+                auto* mg_entry = objects_.Find(
+                    {"system", "snapshots", mg_hash});
+                if (mg_entry) mgs.push_back(mg_entry);
+            }
+            if (page.size() < kPageSize) break;
+            offset += page.size();
+        }
+
+        for (auto* mg : mgs) Unpin(mg);
     }
 }
diff --git a/src/Merkle.cpp b/src/Merkle.cpp
index 56ab349..760cfce 100644
--- a/src/Merkle.cpp
+++ b/src/Merkle.cpp
@@ -3,7 +3,6 @@
 #include <algorithm>
 #include <cassert>
 #include <cstring>
-#include <numeric>
 #include <stdexcept>
 
 namespace nt {
@@ -49,9 +48,9 @@ static uint64_t read_u64(const uint8_t* p)
          | (uint64_t(p[6]) <<  8) |  uint64_t(p[7]);
 }
 
-// ── Hex / binary conversion ──────────────────────────────────────────────────
+// ── Hex / binary conversion (free, key-agnostic) ─────────────────────────────
 
-Merkle::Hash32 Merkle::hex_to_bin(const std::string& hex)
+Hash32 hex_to_bin(const std::string& hex)
 {
     if (hex.size() != 64)
         throw std::invalid_argument(
@@ -70,7 +69,7 @@ Merkle::Hash32 Merkle::hex_to_bin(const std::string& hex)
     return out;
 }
 
-std::string Merkle::bin_to_hex(const Hash32& bin)
+std::string bin_to_hex(const Hash32& bin)
 {
     static const char HEX[] = "0123456789abcdef";
     std::string out(64, '0');
@@ -82,76 +81,136 @@ std::string Merkle::bin_to_hex(const Hash32& bin)
     return out;
 }
 
+// ── Key codec: per-Key serialisation primitives ──────────────────────────────
+
+namespace {
+
+template<typename Key>
+struct KeyCodec; // primary template intentionally undefined
+
+template<>
+struct KeyCodec<Hash32> {
+    static void write(std::vector<uint8_t>& buf, const Hash32& k) {
+        buf.insert(buf.end(), k.begin(), k.end());
+    }
+    // Returns key and advances *p_inout by the number of consumed bytes.
+    static Hash32 read(const uint8_t*& p) {
+        Hash32 k;
+        std::memcpy(k.data(), p, 32);
+        p += 32;
+        return k;
+    }
+    static size_t encoded_size(const Hash32&) { return 32; }
+};
+
+template<>
+struct KeyCodec<std::string> {
+    static void write(std::vector<uint8_t>& buf, const std::string& k) {
+        write_u32(buf, static_cast<uint32_t>(k.size()));
+        buf.insert(buf.end(), k.begin(), k.end());
+    }
+    static std::string read(const uint8_t*& p) {
+        uint32_t len = read_u32(p);
+        p += 4;
+        std::string k(reinterpret_cast<const char*>(p), len);
+        p += len;
+        return k;
+    }
+    static size_t encoded_size(const std::string& k) { return 4 + k.size(); }
+};
+
+} // anonymous namespace
+
 // ── Node serialisation ───────────────────────────────────────────────────────
 
-std::vector<uint8_t> Merkle::encode_leaf(const LeafNode& n)
+template<typename Key>
+std::vector<uint8_t> Merkle<Key>::encode_leaf(const LeafNode& n)
 {
     std::vector<uint8_t> buf;
-    buf.reserve(1 + 4 + n.hashes.size() * 32);
+    size_t payload_size = 0;
+    for (const auto& e : n.entries)
+        payload_size += KeyCodec<Key>::encoded_size(e.key) + 32;
+    buf.reserve(1 + 4 + payload_size);
+
     buf.push_back(TAG_LEAF);
-    write_u32(buf, static_cast<uint32_t>(n.hashes.size()));
-    for (const auto& h : n.hashes)
-        buf.insert(buf.end(), h.begin(), h.end());
+    write_u32(buf, static_cast<uint32_t>(n.entries.size()));
+    for (const auto& e : n.entries)
+    {
+        KeyCodec<Key>::write(buf, e.key);
+        buf.insert(buf.end(), e.payload.begin(), e.payload.end());
+    }
     return buf;
 }
 
-std::vector<uint8_t> Merkle::encode_internal(const InternalNode& n)
+template<typename Key>
+std::vector<uint8_t> Merkle<Key>::encode_internal(const InternalNode& n)
 {
-    // Each entry: 8 (leaf_count) + 32 (min_hash) + 32 (child_hash) = 72 bytes
     std::vector<uint8_t> buf;
-    buf.reserve(1 + 4 + n.entries.size() * 72);
+    size_t payload_size = 0;
+    for (const auto& e : n.entries)
+        payload_size += 8 + KeyCodec<Key>::encoded_size(e.min_key) + 32;
+    buf.reserve(1 + 4 + payload_size);
+
     buf.push_back(TAG_INTERNAL);
     write_u32(buf, static_cast<uint32_t>(n.entries.size()));
     for (const auto& e : n.entries)
     {
         write_u64(buf, e.leaf_count);
-        buf.insert(buf.end(), e.min_hash.begin(), e.min_hash.end());
+        KeyCodec<Key>::write(buf, e.min_key);
         buf.insert(buf.end(), e.child_hash.begin(), e.child_hash.end());
     }
     return buf;
 }
 
-Merkle::LeafNode Merkle::decode_leaf(const std::vector<uint8_t>& bytes)
+template<typename Key>
+typename Merkle<Key>::LeafNode Merkle<Key>::decode_leaf(const std::vector<uint8_t>& bytes)
 {
     uint32_t count = read_u32(bytes.data() + 1);
     LeafNode n;
-    n.hashes.reserve(count);
+    n.entries.reserve(count);
     const uint8_t* p = bytes.data() + 5;
-    for (uint32_t i = 0; i < count; ++i, p += 32)
+    for (uint32_t i = 0; i < count; ++i)
     {
-        Hash32 h;
-        std::memcpy(h.data(), p, 32);
-        n.hashes.push_back(h);
+        LeafEntry e;
+        e.key = KeyCodec<Key>::read(p);
+        std::memcpy(e.payload.data(), p, 32);
+        p += 32;
+        n.entries.push_back(std::move(e));
     }
     return n;
 }
 
-Merkle::InternalNode Merkle::decode_internal(const std::vector<uint8_t>& bytes)
+template<typename Key>
+typename Merkle<Key>::InternalNode Merkle<Key>::decode_internal(const std::vector<uint8_t>& bytes)
 {
     uint32_t count = read_u32(bytes.data() + 1);
     InternalNode n;
     n.entries.reserve(count);
     const uint8_t* p = bytes.data() + 5;
-    for (uint32_t i = 0; i < count; ++i, p += 72)
+    for (uint32_t i = 0; i < count; ++i)
     {
         ChildEntry e;
         e.leaf_count = read_u64(p);
-        std::memcpy(e.min_hash.data(),   p + 8,  32);
-        std::memcpy(e.child_hash.data(), p + 40, 32);
-        n.entries.push_back(e);
+        p += 8;
+        e.min_key = KeyCodec<Key>::read(p);
+        std::memcpy(e.child_hash.data(), p, 32);
+        p += 32;
+        n.entries.push_back(std::move(e));
     }
     return n;
 }
 
-bool Merkle::is_leaf_bytes(const std::vector<uint8_t>& bytes)
+template<typename Key>
+bool Merkle<Key>::is_leaf_bytes(const std::vector<uint8_t>& bytes)
 {
     return !bytes.empty() && bytes[0] == TAG_LEAF;
 }
 
 // ── Storage helpers ──────────────────────────────────────────────────────────
 
-std::vector<uint8_t> Merkle::load_node(IStorageBackend& store,
-                                        const std::string& hash_hex)
+template<typename Key>
+std::vector<uint8_t> Merkle<Key>::load_node(IStorageBackend& store,
+                                            const std::string& hash_hex)
 {
     auto opt = store.Get(hash_hex);
     if (!opt)
@@ -159,46 +218,50 @@ std::vector<uint8_t> Merkle::load_node(IStorageBackend& store,
     return std::move(*opt);
 }
 
-std::string Merkle::store_leaf(IStorageBackend& store, const LeafNode& n)
+template<typename Key>
+std::string Merkle<Key>::store_leaf(IStorageBackend& store, const LeafNode& n)
 {
     return store.Put(encode_leaf(n));
 }
 
-std::string Merkle::store_internal(IStorageBackend& store, const InternalNode& n)
+template<typename Key>
+std::string Merkle<Key>::store_internal(IStorageBackend& store, const InternalNode& n)
 {
     return store.Put(encode_internal(n));
 }
 
-Merkle::Hash32 Merkle::subtree_min(IStorageBackend& store, const std::string& node_hex)
+template<typename Key>
+Key Merkle<Key>::subtree_min_key(IStorageBackend& store, const std::string& node_hex)
 {
     auto bytes = load_node(store, node_hex);
     if (is_leaf_bytes(bytes))
     {
         auto leaf = decode_leaf(bytes);
-        assert(!leaf.hashes.empty() && "Merkle invariant: leaf node must never be empty");
-        return leaf.hashes.front();
+        assert(!leaf.entries.empty() && "Merkle invariant: leaf node must never be empty");
+        return leaf.entries.front().key;
     }
     auto node = decode_internal(bytes);
     assert(!node.entries.empty() && "Merkle invariant: internal node must never be empty");
-    return node.entries.front().min_hash;
+    return node.entries.front().min_key;
 }
 
 // ── Routing helper ───────────────────────────────────────────────────────────
 
-size_t Merkle::route(const InternalNode& node, const Hash32& target)
+template<typename Key>
+size_t Merkle<Key>::route(const InternalNode& node, const Key& target)
 {
     assert(!node.entries.empty() && "Merkle invariant: internal node must never be empty");
     assert(std::is_sorted(node.entries.begin(), node.entries.end(),
                [](const ChildEntry& a, const ChildEntry& b) {
-                   return a.min_hash < b.min_hash;
-               }) && "Merkle invariant: internal node entries must be sorted by min_hash");
+                   return a.min_key < b.min_key;
+               }) && "Merkle invariant: internal node entries must be sorted by min_key");
 
-    // Rightmost child whose min_hash ≤ target.
-    // Defaults to 0 (leftmost) when target < all min_hashes.
+    // Rightmost child whose min_key ≤ target.
+    // Defaults to 0 (leftmost) when target < all min_keys.
     size_t idx = 0;
     for (size_t i = 1; i < node.entries.size(); ++i)
     {
-        if (node.entries[i].min_hash <= target)
+        if (!(target < node.entries[i].min_key))
             idx = i;
         else
             break;
@@ -208,47 +271,48 @@ size_t Merkle::route(const InternalNode& node, const Hash32& target)
 
 // ── Insert ───────────────────────────────────────────────────────────────────
 
-std::string Merkle::Insert(IStorageBackend& store,
-                             const std::string& root_hex,
-                             const std::string& hash_hex)
+template<typename Key>
+std::string Merkle<Key>::Insert(IStorageBackend& store,
+                                 const std::string& root_hex,
+                                 const Key& key,
+                                 const Hash32& payload)
 {
-    const Hash32 tuple_hash = hex_to_bin(hash_hex);
-
     if (root_hex.empty())
     {
         LeafNode leaf;
-        leaf.hashes.push_back(tuple_hash);
+        leaf.entries.push_back({ key, payload });
         return store_leaf(store, leaf);
     }
 
-    auto result = insert_into(store, root_hex, tuple_hash);
+    auto result = insert_into(store, root_hex, key, payload);
     if (!result.did_split)
         return result.new_hash;
 
     // Root split: wrap both halves in a new internal root.
-    Hash32 left_min  = subtree_min(store, result.new_hash);
-    Hash32 right_min = result.split_min_hash;
+    Key left_min  = subtree_min_key(store, result.new_hash);
 
     InternalNode new_root;
 
     ChildEntry left;
     left.leaf_count = result.leaf_count;
-    left.min_hash   = left_min;
+    left.min_key    = left_min;
     left.child_hash = hex_to_bin(result.new_hash);
-    new_root.entries.push_back(left);
+    new_root.entries.push_back(std::move(left));
 
     ChildEntry right;
     right.leaf_count = result.split_leaf_count;
-    right.min_hash   = right_min;
+    right.min_key    = result.split_min_key;
     right.child_hash = hex_to_bin(result.split_hash);
-    new_root.entries.push_back(right);
+    new_root.entries.push_back(std::move(right));
 
     return store_internal(store, new_root);
 }
 
-Merkle::InsertResult Merkle::insert_into(IStorageBackend& store,
-                                           const std::string& node_hex,
-                                           const Hash32& tuple_hash)
+template<typename Key>
+typename Merkle<Key>::InsertResult Merkle<Key>::insert_into(IStorageBackend& store,
+                                                             const std::string& node_hex,
+                                                             const Key& key,
+                                                             const Hash32& payload)
 {
     auto bytes = load_node(store, node_hex);
 
@@ -257,38 +321,51 @@ Merkle::InsertResult Merkle::insert_into(IStorageBackend& store,
     {
         auto leaf = decode_leaf(bytes);
 
-        auto pos = std::lower_bound(leaf.hashes.begin(), leaf.hashes.end(), tuple_hash);
-        if (pos != leaf.hashes.end() && *pos == tuple_hash)
+        auto pos = std::lower_bound(leaf.entries.begin(), leaf.entries.end(), key,
+                                    [](const LeafEntry& e, const Key& k) {
+                                        return e.key < k;
+                                    });
+        if (pos != leaf.entries.end() && !(key < pos->key))
         {
-            // Already present — no-op.
+            // Key already present — update payload if it changed, else no-op.
+            if (pos->payload == payload)
+            {
+                InsertResult r;
+                r.new_hash   = node_hex;
+                r.leaf_count = leaf.entries.size();
+                return r;
+            }
+            pos->payload = payload;
             InsertResult r;
-            r.new_hash   = node_hex;
-            r.leaf_count = leaf.hashes.size();
+            r.new_hash   = store_leaf(store, leaf);
+            r.leaf_count = leaf.entries.size();
             return r;
         }
 
-        leaf.hashes.insert(pos, tuple_hash);
+        leaf.entries.insert(pos, LeafEntry{ key, payload });
 
-        if (leaf.hashes.size() <= B)
+        if (leaf.entries.size() <= B)
         {
             InsertResult r;
             r.new_hash   = store_leaf(store, leaf);
-            r.leaf_count = leaf.hashes.size();
+            r.leaf_count = leaf.entries.size();
             return r;
         }
 
         // Split leaf at midpoint.
-        size_t mid = leaf.hashes.size() / 2;
+        size_t mid = leaf.entries.size() / 2;
         LeafNode left_leaf, right_leaf;
-        left_leaf.hashes  = { leaf.hashes.begin(), leaf.hashes.begin() + static_cast<ptrdiff_t>(mid) };
-        right_leaf.hashes = { leaf.hashes.begin() + static_cast<ptrdiff_t>(mid), leaf.hashes.end() };
+        left_leaf.entries  = { leaf.entries.begin(),
+                               leaf.entries.begin() + static_cast<ptrdiff_t>(mid) };
+        right_leaf.entries = { leaf.entries.begin() + static_cast<ptrdiff_t>(mid),
+                               leaf.entries.end() };
 
         InsertResult r;
         r.new_hash         = store_leaf(store, left_leaf);
-        r.leaf_count       = left_leaf.hashes.size();
+        r.leaf_count       = left_leaf.entries.size();
         r.did_split        = true;
-        r.split_leaf_count = right_leaf.hashes.size();
-        r.split_min_hash   = right_leaf.hashes.front();
+        r.split_leaf_count = right_leaf.entries.size();
+        r.split_min_key    = right_leaf.entries.front().key;
         r.split_hash       = store_leaf(store, right_leaf);
         return r;
     }
@@ -296,19 +373,20 @@ Merkle::InsertResult Merkle::insert_into(IStorageBackend& store,
     // ── Internal node ───────────────────────────────────────────────────────
     auto node = decode_internal(bytes);
 
-    size_t child_idx = route(node, tuple_hash);
+    size_t child_idx = route(node, key);
     auto child_result = insert_into(store,
                                      bin_to_hex(node.entries[child_idx].child_hash),
-                                     tuple_hash);
+                                     key,
+                                     payload);
 
     node.entries[child_idx].leaf_count = child_result.leaf_count;
     node.entries[child_idx].child_hash = hex_to_bin(child_result.new_hash);
 
-    // Update min_hash for the modified child — only child 0 can have its min
-    // decrease (routing guarantees min_hash[i>0] ≤ inserted hash, so inserting
-    // below min_hash[i>0] is impossible).
+    // Update min_key for the modified child — only child 0 can have its min
+    // decrease (routing guarantees min_key[i>0] ≤ inserted key, so inserting
+    // below min_key[i>0] is impossible).
     if (child_idx == 0)
-        node.entries[0].min_hash = subtree_min(store, child_result.new_hash);
+        node.entries[0].min_key = subtree_min_key(store, child_result.new_hash);
 
     uint64_t total = 0;
     for (const auto& e : node.entries) total += e.leaf_count;
@@ -324,9 +402,10 @@ Merkle::InsertResult Merkle::insert_into(IStorageBackend& store,
     // Insert the right sibling immediately after child_idx.
     ChildEntry sibling;
     sibling.leaf_count = child_result.split_leaf_count;
-    sibling.min_hash   = child_result.split_min_hash;
+    sibling.min_key    = child_result.split_min_key;
     sibling.child_hash = hex_to_bin(child_result.split_hash);
-    node.entries.insert(node.entries.begin() + static_cast<ptrdiff_t>(child_idx) + 1, sibling);
+    node.entries.insert(node.entries.begin() + static_cast<ptrdiff_t>(child_idx) + 1,
+                        std::move(sibling));
     total += child_result.split_leaf_count;
 
     if (node.entries.size() <= B)
@@ -340,8 +419,10 @@ Merkle::InsertResult Merkle::insert_into(IStorageBackend& store,
     // Split this internal node.
     size_t mid = node.entries.size() / 2;
     InternalNode left_node, right_node;
-    left_node.entries  = { node.entries.begin(), node.entries.begin() + static_cast<ptrdiff_t>(mid) };
-    right_node.entries = { node.entries.begin() + static_cast<ptrdiff_t>(mid), node.entries.end() };
+    left_node.entries  = { node.entries.begin(),
+                           node.entries.begin() + static_cast<ptrdiff_t>(mid) };
+    right_node.entries = { node.entries.begin() + static_cast<ptrdiff_t>(mid),
+                           node.entries.end() };
 
     uint64_t left_count = 0, right_count = 0;
     for (const auto& e : left_node.entries)  left_count  += e.leaf_count;
@@ -352,20 +433,20 @@ Merkle::InsertResult Merkle::insert_into(IStorageBackend& store,
     r.leaf_count       = left_count;
     r.did_split        = true;
     r.split_leaf_count = right_count;
-    r.split_min_hash   = right_node.entries.front().min_hash;
+    r.split_min_key    = right_node.entries.front().min_key;
     r.split_hash       = store_internal(store, right_node);
     return r;
 }
 
 // ── Remove ───────────────────────────────────────────────────────────────────
 
-std::string Merkle::Remove(IStorageBackend& store,
-                             const std::string& root_hex,
-                             const std::string& hash_hex)
+template<typename Key>
+std::string Merkle<Key>::Remove(IStorageBackend& store,
+                                 const std::string& root_hex,
+                                 const Key& key)
 {
     if (root_hex.empty()) return "";
-    const Hash32 tuple_hash = hex_to_bin(hash_hex);
-    auto result = remove_from(store, root_hex, tuple_hash);
+    auto result = remove_from(store, root_hex, key);
     if (result.new_hash.empty()) return "";
 
     // Collapse: if the root is an internal node with a single child, skip it.
@@ -379,9 +460,10 @@ std::string Merkle::Remove(IStorageBackend& store,
     return result.new_hash;
 }
 
-Merkle::RemoveResult Merkle::remove_from(IStorageBackend& store,
-                                           const std::string& node_hex,
-                                           const Hash32& tuple_hash)
+template<typename Key>
+typename Merkle<Key>::RemoveResult Merkle<Key>::remove_from(IStorageBackend& store,
+                                                             const std::string& node_hex,
+                                                             const Key& key)
 {
     auto bytes = load_node(store, node_hex);
 
@@ -389,23 +471,26 @@ Merkle::RemoveResult Merkle::remove_from(IStorageBackend& store,
     if (is_leaf_bytes(bytes))
     {
         auto leaf = decode_leaf(bytes);
-        auto it = std::lower_bound(leaf.hashes.begin(), leaf.hashes.end(), tuple_hash);
-        if (it == leaf.hashes.end() || *it != tuple_hash)
-            return { node_hex, leaf.hashes.size() };  // not found — no change
-
-        leaf.hashes.erase(it);
-        if (leaf.hashes.empty())
+        auto it = std::lower_bound(leaf.entries.begin(), leaf.entries.end(), key,
+                                   [](const LeafEntry& e, const Key& k) {
+                                       return e.key < k;
+                                   });
+        if (it == leaf.entries.end() || key < it->key)
+            return { node_hex, leaf.entries.size() };  // not found — no change
+
+        leaf.entries.erase(it);
+        if (leaf.entries.empty())
             return { "", 0 };
-        return { store_leaf(store, leaf), leaf.hashes.size() };
+        return { store_leaf(store, leaf), leaf.entries.size() };
     }
 
     // ── Internal node ───────────────────────────────────────────────────────
     auto node = decode_internal(bytes);
 
-    size_t child_idx = route(node, tuple_hash);
+    size_t child_idx = route(node, key);
     auto child_result = remove_from(store,
                                      bin_to_hex(node.entries[child_idx].child_hash),
-                                     tuple_hash);
+                                     key);
 
     if (child_result.new_hash.empty())
     {
@@ -417,7 +502,7 @@ Merkle::RemoveResult Merkle::remove_from(IStorageBackend& store,
         node.entries[child_idx].leaf_count = child_result.leaf_count;
         node.entries[child_idx].child_hash = hex_to_bin(child_result.new_hash);
         // The min of any child might have changed if we removed its minimum entry.
-        node.entries[child_idx].min_hash   = subtree_min(store, child_result.new_hash);
+        node.entries[child_idx].min_key    = subtree_min_key(store, child_result.new_hash);
     }
 
     uint64_t total = 0;
@@ -425,36 +510,68 @@ Merkle::RemoveResult Merkle::remove_from(IStorageBackend& store,
     return { store_internal(store, node), total };
 }
 
+// ── Get ──────────────────────────────────────────────────────────────────────
+
+template<typename Key>
+std::optional<Hash32> Merkle<Key>::Get(IStorageBackend& store,
+                                        const std::string& root_hex,
+                                        const Key& key)
+{
+    if (root_hex.empty()) return std::nullopt;
+    return get_from(store, root_hex, key);
+}
+
+template<typename Key>
+std::optional<Hash32> Merkle<Key>::get_from(IStorageBackend& store,
+                                             const std::string& node_hex,
+                                             const Key& key)
+{
+    auto bytes = load_node(store, node_hex);
+    if (is_leaf_bytes(bytes))
+    {
+        auto leaf = decode_leaf(bytes);
+        auto it = std::lower_bound(leaf.entries.begin(), leaf.entries.end(), key,
+                                   [](const LeafEntry& e, const Key& k) {
+                                       return e.key < k;
+                                   });
+        if (it == leaf.entries.end() || key < it->key)
+            return std::nullopt;
+        return it->payload;
+    }
+    auto node = decode_internal(bytes);
+    size_t child_idx = route(node, key);
+    return get_from(store, bin_to_hex(node.entries[child_idx].child_hash), key);
+}
+
 // ── Page ─────────────────────────────────────────────────────────────────────
 
-std::vector<std::string> Merkle::Page(IStorageBackend& store,
-                                       const std::string& root_hex,
-                                       size_t offset,
-                                       size_t limit)
+template<typename Key>
+std::vector<typename Merkle<Key>::Entry> Merkle<Key>::Page(IStorageBackend& store,
+                                                            const std::string& root_hex,
+                                                            size_t offset,
+                                                            size_t limit)
 {
     if (root_hex.empty() || limit == 0) return {};
-    std::vector<Hash32> raw;
-    raw.reserve(limit);
+    std::vector<Entry> out;
+    out.reserve(limit);
     size_t skip = offset;
-    page_from(store, root_hex, skip, limit, raw);
-    std::vector<std::string> out;
-    out.reserve(raw.size());
-    for (const auto& h : raw) out.push_back(bin_to_hex(h));
+    page_from(store, root_hex, skip, limit, out);
     return out;
 }
 
-void Merkle::page_from(IStorageBackend& store,
-                        const std::string& node_hex,
-                        size_t& offset,
-                        size_t limit,
-                        std::vector<Hash32>& out)
+template<typename Key>
+void Merkle<Key>::page_from(IStorageBackend& store,
+                             const std::string& node_hex,
+                             size_t& offset,
+                             size_t limit,
+                             std::vector<Entry>& out)
 {
     auto bytes = load_node(store, node_hex);
 
     if (is_leaf_bytes(bytes))
     {
         auto leaf = decode_leaf(bytes);
-        const size_t available = leaf.hashes.size();
+        const size_t available = leaf.entries.size();
 
         if (offset >= available)
         {
@@ -463,7 +580,7 @@ void Merkle::page_from(IStorageBackend& store,
         }
 
         for (size_t i = offset; i < available && out.size() < limit; ++i)
-            out.push_back(leaf.hashes[i]);
+            out.push_back(Entry{ leaf.entries[i].key, leaf.entries[i].payload });
 
         offset = 0;
         return;
@@ -485,4 +602,9 @@ void Merkle::page_from(IStorageBackend& store,
     }
 }
 
+// ── Explicit instantiations ──────────────────────────────────────────────────
+
+template class Merkle<Hash32>;
+template class Merkle<std::string>;
+
 } // namespace nt
diff --git a/src/MultigroupCodec.cpp b/src/MultigroupCodec.cpp
new file mode 100644
index 0000000..81d305c
--- /dev/null
+++ b/src/MultigroupCodec.cpp
@@ -0,0 +1,78 @@
+#include "MultigroupCodec.h"
+
+#include "Merkle.h"
+
+namespace nt::MultigroupCodec
+{
+    // A relation with zero tuples reports an empty merkle_root externally.
+    // The merkle tree's payload type is fixed-width Hash32, so the codec
+    // maps "" ↔ the all-zero hash. The collision probability with a real
+    // SHA256 hitting all-zero (2^-256) is treated as unreachable.
+    static nt::Hash32 encode_payload(const std::string& hex)
+    {
+        if (hex.empty()) return nt::Hash32{};
+        return nt::hex_to_bin(hex);
+    }
+
+    static std::string decode_payload(const nt::Hash32& bin)
+    {
+        static const nt::Hash32 zero{};
+        if (bin == zero) return "";
+        return nt::bin_to_hex(bin);
+    }
+
+    std::string Build(IStorageBackend& store,
+                      const std::vector<RelationEntry>& relations)
+    {
+        std::string root;
+        for (const auto& [name, root_hex] : relations)
+            root = nt::Merkle<std::string>::Insert(store, root, name,
+                                                    encode_payload(root_hex));
+        return root;
+    }
+
+    std::vector<RelationEntry> List(IStorageBackend& store,
+                                     const std::string& root_hex)
+    {
+        std::vector<RelationEntry> out;
+        constexpr size_t kPageSize = 1024;
+        size_t offset = 0;
+        while (true)
+        {
+            auto page = nt::Merkle<std::string>::Page(store, root_hex,
+                                                      offset, kPageSize);
+            if (page.empty()) break;
+            for (auto& entry : page)
+                out.emplace_back(std::move(entry.key),
+                                  decode_payload(entry.payload));
+            if (page.size() < kPageSize) break;
+            offset += page.size();
+        }
+        return out;
+    }
+
+    std::string InsertOne(IStorageBackend& store,
+                          const std::string& root_hex,
+                          const std::string& relation_name,
+                          const std::string& relation_root_hex)
+    {
+        return nt::Merkle<std::string>::Insert(store, root_hex, relation_name,
+                                                encode_payload(relation_root_hex));
+    }
+
+    std::string RemoveOne(IStorageBackend& store,
+                          const std::string& root_hex,
+                          const std::string& relation_name)
+    {
+        return nt::Merkle<std::string>::Remove(store, root_hex, relation_name);
+    }
+
+    std::string Lookup(IStorageBackend& store,
+                       const std::string& root_hex,
+                       const std::string& relation_name)
+    {
+        auto found = nt::Merkle<std::string>::Get(store, root_hex, relation_name);
+        if (!found) return "";
+        return decode_payload(*found);
+    }
+}
diff --git a/src/NamespaceReferenceManager.cpp b/src/NamespaceReferenceManager.cpp
index 1d9baac..455d077 100644
--- a/src/NamespaceReferenceManager.cpp
+++ b/src/NamespaceReferenceManager.cpp
@@ -1,5 +1,132 @@
 #include "NamespaceReferenceManager.h"
 
+#include "Merkle.h"
+
 namespace nt
 {
+    NamespaceReferenceManager::NamespaceReferenceManager(ObjectManager& objects,
+                                                          IStorageBackend& storage)
+        : objects_(objects), storage_(storage)
+    {}
+
+    // Resolves the multigroup hash that `mg_name` currently maps to inside the
+    // branch-tree rooted at `branch_tree_root`. Returns the empty string when
+    // the branch is unborn (empty root) or the multigroup is absent.
+    static std::string lookup_mg_hash(IStorageBackend& storage,
+                                       const std::string& branch_tree_root,
+                                       const std::string& mg_name)
+    {
+        if (branch_tree_root.empty()) return "";
+        auto found = Merkle<std::string>::Get(storage, branch_tree_root, mg_name);
+        if (!found) return "";
+        // All-zero payload = "no snapshot bound" sentinel matching MultigroupCodec.
+        static const Hash32 zero{};
+        if (*found == zero) return "";
+        return bin_to_hex(*found);
+    }
+
+    std::vector<std::string> NamespaceReferenceManager::Resolve(
+        std::vector<std::string> path) const
+    {
+        constexpr int kMaxReparseDepth = 30;
+
+        for (int depth = 0; depth < kMaxReparseDepth; ++depth)
+        {
+            // /system/sessions/<X>/branches/<name>/multigroups/<mg>/<sub>...
+            // — per-session override takes precedence; absent override falls
+            // back to the global branch. Requires at minimum the literal
+            // "multigroups" qualifier and an mg name.
+            if (path.size() > 6
+                && path[0] == "system"
+                && path[1] == "sessions"
+                && path[3] == "branches"
+                && path[5] == "multigroups")
+            {
+                const std::vector<std::string> session_path{
+                    "system", "sessions", path[2]
+                };
+                auto* session_entry = objects_.Find(session_path);
+
+                std::string branch_tree_root;
+                if (session_entry && session_entry->object)
+                {
+                    auto* session = dynamic_cast<ObjectManager::Session*>(
+                        session_entry->object.get());
+                    if (session)
+                    {
+                        auto it = session->branch_overrides.find(path[4]);
+                        if (it != session->branch_overrides.end())
+                            branch_tree_root = it->second;
+                    }
+                }
+
+                // No session-level override — defer to the global branch's HEAD.
+                if (branch_tree_root.empty())
+                {
+                    const std::vector<std::string> branch_path{
+                        "system", "branches", path[4]
+                    };
+                    auto* bentry = objects_.Find(branch_path);
+                    if (bentry && bentry->object)
+                    {
+                        auto* branch = dynamic_cast<ObjectManager::Branch*>(
+                            bentry->object.get());
+                        if (branch) branch_tree_root = branch->target_hash;
+                    }
+                }
+
+                const std::string mg_hash =
+                    lookup_mg_hash(storage_, branch_tree_root, path[6]);
+                if (mg_hash.empty()) return path;
+
+                std::vector<std::string> rewritten{
+                    "system", "snapshots", mg_hash
+                };
+                for (size_t i = 7; i < path.size(); ++i)
+                    rewritten.push_back(path[i]);
+
+                path = std::move(rewritten);
+                continue;
+            }
+
+            // /system/branches/<name>/multigroups/<mg>/<sub>... →
+            //   /system/snapshots/<mg_hash>/<sub>...
+            // Requires the literal "multigroups" qualifier and an mg name. A
+            // bare /system/branches/<name> path opens the branch object itself
+            // and is returned unchanged.
+            if (path.size() > 4
+                && path[0] == "system"
+                && path[1] == "branches"
+                && path[3] == "multigroups")
+            {
+                const std::vector<std::string> branch_path{
+                    "system", "branches", path[2]
+                };
+                auto* entry = objects_.Find(branch_path);
+                if (!entry || !entry->object) return path;
+
+                auto* branch =
+                    dynamic_cast<ObjectManager::Branch*>(entry->object.get());
+                if (!branch) return path;
+
+                const std::string mg_hash =
+                    lookup_mg_hash(storage_, branch->target_hash, path[4]);
+                if (mg_hash.empty()) return path;
+
+                std::vector<std::string> rewritten{
+                    "system", "snapshots", mg_hash
+                };
+                for (size_t i = 5; i < path.size(); ++i)
+                    rewritten.push_back(path[i]);
+
+                path = std::move(rewritten);
+                continue;
+            }
+
+            return path;
+        }
+
+        // Cycle guard tripped — surface as a lookup miss.
+        return {};
+    }
 }
diff --git a/src/ObjectManager.cpp b/src/ObjectManager.cpp
index 19f862d..67504c3 100644
--- a/src/ObjectManager.cpp
+++ b/src/ObjectManager.cpp
@@ -30,4 +30,19 @@ namespace nt
         }
         return nullptr;
     }
+
+    bool ObjectManager::Unregister(const std::vector<std::string>& object_path)
+    {
+        std::unique_ptr<registry>* slot = &entries;
+        while (slot->get() != nullptr)
+        {
+            if ((*slot)->head->path == object_path)
+            {
+                *slot = std::move((*slot)->next);
+                return true;
+            }
+            slot = &(*slot)->next;
+        }
+        return false;
+    }
 }
diff --git a/src/RNT_C_API.cpp b/src/RNT_C_API.cpp
index 34bbacc..c92d03b 100644
--- a/src/RNT_C_API.cpp
+++ b/src/RNT_C_API.cpp
@@ -5,6 +5,8 @@
 #include "IdentityManager.h"
 #include "LifecycleManager.h"
 #include "Merkle.h"
+#include "MultigroupCodec.h"
+#include "NamespaceReferenceManager.h"
 #include "ObjectManager.h"
 #include "PermissionsManager.h"
 #include "InMemoryBackend.h"
@@ -12,9 +14,11 @@
 #include "TupleCodec.h"
 #include "VM.h"
 
+#include <algorithm>
 #include <cstring>
 #include <memory>
 #include <mutex>
+#include <random>
 #include <sstream>
 #include <stdexcept>
 #include <string>
@@ -38,16 +42,16 @@ namespace
         nt::ObjectManager      objects;
         nt::PermissionsManager permissions;
         nt::IdentityManager    identities;
-        nt::LifecycleManager   lifecycles;
+        std::unique_ptr<nt::LifecycleManager> lifecycles;
+        std::unique_ptr<nt::NamespaceReferenceManager> references;
         std::unique_ptr<nt::HandlerManager> handler;
         std::unique_ptr<nt::CursorManager>  cursors;
     };
 
     static std::unique_ptr<Runtime> g_rt;
-    static std::mutex               g_init_mutex;
     static bool                     g_init_done = false;
 
-    bool is_init() { return g_init_done; }
+    bool is_initialized() { return g_init_done; }
 
     static std::vector<std::string> split_path(const char* path)
     {
@@ -121,6 +125,286 @@ namespace
         return t;
     }
 
+    static std::unique_ptr<nt::ObjectManager::object_type> make_multigroup_type()
+    {
+        auto t      = std::make_unique<nt::ObjectManager::object_type>();
+        t->label    = MULTIGROUP;
+        t->disposable = false;
+        t->methods  = { OPEN, CLOSE };
+        // Multigroup snapshots are immutable and content-addressed; concurrent
+        // readers never contend. GC is gated by reference_count only.
+        t->exclusive = false;
+        return t;
+    }
+
+    static std::unique_ptr<nt::ObjectManager::object_type> make_branch_tree_type()
+    {
+        auto t      = std::make_unique<nt::ObjectManager::object_type>();
+        t->label    = BRANCH_TREE;
+        t->disposable = false;
+        t->methods  = { OPEN, CLOSE };
+        // Branch-tree blobs are immutable content-addressed merkle roots;
+        // GC is gated by reference_count only.
+        t->exclusive = false;
+        return t;
+    }
+
+    static std::unique_ptr<nt::ObjectManager::object_type> make_session_type()
+    {
+        auto t      = std::make_unique<nt::ObjectManager::object_type>();
+        t->label    = SESSION;
+        // Sessions live only as long as their connection; rnt_session_close
+        // removes them via ObjectManager::Unregister.
+        t->disposable = true;
+        t->methods  = { OPEN, CLOSE };
+        // Multiple readers/writers may consult a session concurrently to read
+        // overrides; contention is only enforced when an exclusive mutator
+        // (set_branch) is wired up. Default to non-exclusive for now.
+        t->exclusive = false;
+        return t;
+    }
+
+    // Generates a random 256-bit hex hash for naming a session. Hashes are
+    // drawn from std::random_device-seeded mt19937_64; we do not need
+    // cryptographic strength here, just enough entropy that two concurrent
+    // rnt_session_open calls cannot collide.
+    static std::string random_session_hash()
+    {
+        static thread_local std::mt19937_64 rng{ std::random_device{}() };
+        static const char hex[] = "0123456789abcdef";
+
+        std::string out;
+        out.reserve(64);
+        for (int i = 0; i < 4; ++i)
+        {
+            uint64_t v = rng();
+            for (int b = 0; b < 8; ++b)
+            {
+                uint8_t byte = static_cast<uint8_t>(v & 0xFF);
+                out.push_back(hex[(byte >> 4) & 0xF]);
+                out.push_back(hex[byte & 0xF]);
+                v >>= 8;
+            }
+        }
+        return out;
+    }
+
+    // Serializes the (name, merkle_root) pairs into the storage backend and
+    // registers the resulting multigroup snapshot under /system/snapshots/<hash>,
+    // with one Relation entry per child at /system/snapshots/<hash>/<name>.
+    // Idempotent: if the snapshot already exists in the registry the existing
+    // entries are reused. Returns the multigroup hash, or empty string on error.
+    //
+    // Each child Relation is Pinned by its parent Multigroup at creation time.
+    // LifecycleManager::CascadeMultigroup releases these pins when the
+    // multigroup itself becomes GC-eligible.
+    static std::string register_snapshot(
+        const std::vector<nt::MultigroupCodec::RelationEntry>& relations)
+    {
+        if (!g_rt) return {};
+
+        const std::string hash = nt::MultigroupCodec::Build(*g_rt->storage, relations);
+        if (hash.empty()) return {};
+        const auto snap_path = split_path(("/system/snapshots/" + hash).c_str());
+
+        if (g_rt->objects.Find(snap_path) != nullptr) return hash;
+
+        auto mg = std::make_unique<nt::ObjectManager::Multigroup>();
+        mg->merkle_root = hash;
+        g_rt->objects.Register(snap_path, std::move(mg), make_multigroup_type());
+
+        for (const auto& [name, root] : relations)
+        {
+            const auto rel_path = split_path(
+                ("/system/snapshots/" + hash + "/relations/" + name).c_str());
+            if (g_rt->objects.Find(rel_path) != nullptr) continue;
+
+            auto rel = std::make_unique<nt::ObjectManager::Relation>();
+            rel->merkle_root = root;
+            g_rt->objects.Register(rel_path, std::move(rel), make_relation_type());
+
+            // Structural pin from the parent Multigroup; released by
+            // LifecycleManager::CascadeMultigroup at GC time.
+            g_rt->lifecycles->Pin(g_rt->objects.Find(rel_path));
+        }
+        return hash;
+    }
+
+    // Enumerates (mg_name, mg_hash) pairs in a branch-tree.
+    static std::vector<std::pair<std::string, std::string>>
+    list_branch_tree(const std::string& branch_tree_root)
+    {
+        std::vector<std::pair<std::string, std::string>> out;
+        if (branch_tree_root.empty()) return out;
+        constexpr size_t kPageSize = 1024;
+        size_t offset = 0;
+        while (true)
+        {
+            auto page = nt::Merkle<std::string>::Page(*g_rt->storage,
+                                                      branch_tree_root,
+                                                      offset, kPageSize);
+            if (page.empty()) break;
+            for (auto& entry : page)
+                out.emplace_back(std::move(entry.key),
+                                  nt::bin_to_hex(entry.payload));
+            if (page.size() < kPageSize) break;
+            offset += page.size();
+        }
+        return out;
+    }
+
+    // Registers /system/branch_trees/<hash> as a BRANCH_TREE if missing, and
+    // takes one structural pin on each multigroup the tree references. The
+    // pins are mirrored by LifecycleManager::CascadeBranchTree at GC time.
+    // Idempotent: a second call with the same hash returns immediately.
+    static void register_branch_tree(const std::string& tree_root)
+    {
+        if (tree_root.empty()) return;
+        const auto bt_path = split_path(
+            ("/system/branch_trees/" + tree_root).c_str());
+        if (g_rt->objects.Find(bt_path) != nullptr) return;
+
+        auto bt = std::make_unique<nt::ObjectManager::BranchTree>();
+        bt->merkle_root = tree_root;
+        g_rt->objects.Register(bt_path, std::move(bt), make_branch_tree_type());
+
+        for (const auto& [_, mg_hash] : list_branch_tree(tree_root))
+        {
+            if (mg_hash.empty()) continue;
+            const auto p = split_path(("/system/snapshots/" + mg_hash).c_str());
+            g_rt->lifecycles->Pin(g_rt->objects.Find(p));
+        }
+    }
+
+    // Adjusts pins when a branch (or session override) moves its bound
+    // branch-tree root. The new tree is pinned before the old is unpinned so
+    // a no-op advance never sees a pinned BranchTree's ref_count briefly hit
+    // zero and trigger cascade collection of every mg it references.
+    static void rebind_branch_tree_pin(const std::string& old_tree_root,
+                                        const std::string& new_tree_root)
+    {
+        if (old_tree_root == new_tree_root) return;
+        if (!new_tree_root.empty())
+        {
+            register_branch_tree(new_tree_root);
+            const auto p = split_path(
+                ("/system/branch_trees/" + new_tree_root).c_str());
+            g_rt->lifecycles->Pin(g_rt->objects.Find(p));
+        }
+        if (!old_tree_root.empty())
+        {
+            const auto p = split_path(
+                ("/system/branch_trees/" + old_tree_root).c_str());
+            g_rt->lifecycles->Unpin(g_rt->objects.Find(p));
+        }
+    }
+
+    // Looks up the mg_hash currently bound to `mg_name` in the branch's tree.
+    // Returns empty string when the branch is unborn or the mg is absent.
+    static std::string lookup_branch_mg(const nt::ObjectManager::Branch& branch,
+                                         const std::string& mg_name)
+    {
+        if (branch.target_hash.empty()) return "";
+        auto found = nt::Merkle<std::string>::Get(*g_rt->storage,
+                                                   branch.target_hash, mg_name);
+        if (!found) return "";
+        static const nt::Hash32 zero{};
+        if (*found == zero) return "";
+        return nt::bin_to_hex(*found);
+    }
+
+    // Reads the (relation_name, root) list of a specific multigroup under a
+    // branch. Returns empty for unborn branches or absent multigroups.
+    static std::vector<nt::MultigroupCodec::RelationEntry> read_mg_relations(
+        const nt::ObjectManager::Branch& branch,
+        const std::string& mg_name)
+    {
+        const std::string mg_hash = lookup_branch_mg(branch, mg_name);
+        if (mg_hash.empty()) return {};
+        return nt::MultigroupCodec::List(*g_rt->storage, mg_hash);
+    }
+
+    // Looks up a branch object by path. Returns nullptr when the path does not
+    // resolve to a BRANCH.
+    static nt::ObjectManager::Branch* find_branch(
+        const std::vector<std::string>& branch_path)
+    {
+        auto* entry = g_rt->objects.Find(branch_path);
+        if (!entry || !entry->object || !entry->head) return nullptr;
+        if (entry->head->type->label != BRANCH) return nullptr;
+        return dynamic_cast<nt::ObjectManager::Branch*>(entry->object.get());
+    }
+
+    // Parses a path of the form
+    //   /system/branches/<name>/multigroups/<mg>/relations/<rel>.
+    // Returns true on success; out-parameters are populated with the branch
+    // path, the multigroup name, and the relation name. Returns false for
+    // any other path shape.
+    static bool split_branch_mg_relation(const std::vector<std::string>& parts,
+                                          std::vector<std::string>& branch_path_out,
+                                          std::string& mg_name_out,
+                                          std::string& relation_name_out)
+    {
+        if (parts.size() != 7) return false;
+        if (parts[0] != "system"      || parts[1] != "branches"
+         || parts[3] != "multigroups" || parts[5] != "relations") return false;
+        branch_path_out   = { "system", "branches", parts[2] };
+        mg_name_out       = parts[4];
+        relation_name_out = parts[6];
+        return true;
+    }
+
+    // Applies a per-relation root update with the full two-level cascade:
+    //   1. Read the (rel, root) list of `mg_name` from the branch's current
+    //      tree (or empty when the mg is absent / branch is unborn).
+    //   2. Insert/replace (relation_name, new_root) in that list.
+    //   3. register_snapshot → new_mg_hash.
+    //   4. Insert (mg_name → new_mg_hash) into the branch tree → new branch
+    //      tree root.
+    //   5. Advance branch.target_hash to the new root and rebind pins.
+    // Returns the new branch tree root on success; empty on error.
+    //
+    // Pin/unpin transitions are applied via rebind_branch_tree_pin: new mgs
+    // are pinned before old mgs are unpinned, so an unchanged advance never
+    // briefly drops a multigroup's ref_count to zero.
+    static std::string commit_relation_update(
+        nt::ObjectManager::Branch& branch,
+        const std::string& mg_name,
+        const std::string& relation_name,
+        const std::string& new_root)
+    {
+        auto relations = read_mg_relations(branch, mg_name);
+        auto it = std::find_if(relations.begin(), relations.end(),
+                               [&](const auto& e) { return e.first == relation_name; });
+        if (it != relations.end()) it->second = new_root;
+        else relations.emplace_back(relation_name, new_root);
+
+        const std::string new_mg_hash = register_snapshot(relations);
+        if (new_mg_hash.empty()) return {};
+
+        const nt::Hash32 mg_payload = nt::hex_to_bin(new_mg_hash);
+        const std::string new_tree_root = nt::Merkle<std::string>::Insert(
+            *g_rt->storage, branch.target_hash, mg_name, mg_payload);
+
+        const std::string old_tree_root = branch.target_hash;
+        branch.target_hash = new_tree_root;
+        rebind_branch_tree_pin(old_tree_root, new_tree_root);
+        return new_tree_root;
+    }
+
+    // Returns the merkle_root of a named relation in the specified mg under
+    // the branch's current tree. Empty when the branch is unborn, the mg
+    // doesn't exist, or the relation is absent.
+    static std::string read_relation_root(const nt::ObjectManager::Branch& branch,
+                                          const std::string& mg_name,
+                                          const std::string& relation_name)
+    {
+        const auto relations = read_mg_relations(branch, mg_name);
+        auto it = std::find_if(relations.begin(), relations.end(),
+                               [&](const auto& e) { return e.first == relation_name; });
+        return (it == relations.end()) ? std::string{} : it->second;
+    }
+
     // -----------------------------------------------------------------------
     // Plan builder internals
     // -----------------------------------------------------------------------
@@ -187,9 +471,6 @@ namespace
 
 int rnt_init(const char* driver, const char* storage_path)
 {
-    std::lock_guard<std::mutex> lock(g_init_mutex);
-    if (g_init_done) return 0;
-
     try
     {
         auto rt = std::make_unique<Runtime>();
@@ -201,19 +482,25 @@ int rnt_init(const char* driver, const char* storage_path)
             rt->storage = std::make_unique<nt::SqliteBackend>(
                 storage_path ? storage_path : ":memory:");
 
+        rt->lifecycles = std::make_unique<nt::LifecycleManager>(
+            rt->objects, *rt->storage);
+        rt->references = std::make_unique<nt::NamespaceReferenceManager>(
+            rt->objects, *rt->storage);
         rt->handler = std::make_unique<nt::HandlerManager>(
-            rt->objects, rt->permissions, rt->identities, rt->lifecycles);
-        rt->cursors = std::make_unique<nt::CursorManager>(*rt->storage);
+            rt->objects, rt->permissions, rt->identities, *rt->lifecycles,
+            *rt->references);
+        rt->cursors = std::make_unique<nt::CursorManager>(
+            *rt->storage, rt->lifecycles.get(), &rt->objects);
         g_rt = std::move(rt);
 
         // Register the default branch on first boot so that
         // rnt_open_handle("/system/branches/master") always succeeds.
-        const auto main_path = split_path("/system/branches/master");
-        if (g_rt->objects.Find(main_path) == nullptr)
+        const auto master_path = split_path("/system/branches/master");
+        if (g_rt->objects.Find(master_path) == nullptr)
         {
             auto branch  = std::make_unique<nt::ObjectManager::Branch>();
             branch->name = "master";
-            g_rt->objects.Register(main_path, std::move(branch),
+            g_rt->objects.Register(master_path, std::move(branch),
                                    make_branch_type());
         }
 
@@ -229,28 +516,107 @@ int rnt_init(const char* driver, const char* storage_path)
 
 int rnt_firewall(const char* /*auth_method*/, char** claims_out)
 {
-    if (!is_init() || !claims_out) return -1;
+    if (!is_initialized() || !claims_out) return -1;
     // PermissionsManager::Firewall is stubbed; grant all claims for now.
     *claims_out = heap_str("READ WRITE");
     return 0;
 }
 
+int rnt_session_open(void* connection_context, char** session_hash_out)
+{
+    if (!is_initialized() || !session_hash_out) return -1;
+
+    // Retry on the astronomically unlikely event of a hash collision against
+    // an already-registered session.
+    std::string hash;
+    for (int attempts = 0; attempts < 8; ++attempts)
+    {
+        hash = random_session_hash();
+        const auto session_path = split_path(("/system/sessions/" + hash).c_str());
+        if (g_rt->objects.Find(session_path) != nullptr) continue;
+
+        auto session = std::make_unique<nt::ObjectManager::Session>();
+        session->connection_context = connection_context;
+        g_rt->objects.Register(session_path, std::move(session), make_session_type());
+        *session_hash_out = heap_str(hash);
+        return 0;
+    }
+    return -1;
+}
+
+int rnt_session_close(const char* session_hash)
+{
+    if (!is_initialized() || !session_hash) return -1;
+    const auto session_path = split_path(
+        ("/system/sessions/" + std::string(session_hash)).c_str());
+
+    auto* entry = g_rt->objects.Find(session_path);
+    if (!entry || !entry->object) return -1;
+    auto* session = dynamic_cast<nt::ObjectManager::Session*>(entry->object.get());
+    if (!session) return -1;
+
+    // Release every pin this session was holding before the entry disappears,
+    // so the dependent multigroups become GC-eligible.
+    for (const auto& [name, hash] : session->branch_overrides)
+        rebind_branch_tree_pin(hash, "");
+    session->branch_overrides.clear();
+
+    return g_rt->objects.Unregister(session_path) ? 0 : -1;
+}
+
+int rnt_session_set_branch(const char* session_hash,
+                           const char* branch_name,
+                           const char* target_hash)
+{
+    if (!is_initialized() || !session_hash || !branch_name || !target_hash)
+        return -1;
+
+    const auto session_path = split_path(
+        ("/system/sessions/" + std::string(session_hash)).c_str());
+    auto* entry = g_rt->objects.Find(session_path);
+    if (!entry || !entry->object || !entry->head) return -1;
+    if (entry->head->type->label != SESSION) return -1;
+
+    auto* session = dynamic_cast<nt::ObjectManager::Session*>(entry->object.get());
+    if (!session) return -1;
+
+    const std::string hash(target_hash);
+    std::string old_hash;
+    auto it = session->branch_overrides.find(branch_name);
+    if (it != session->branch_overrides.end()) old_hash = it->second;
+
+    if (hash.empty())
+    {
+        session->branch_overrides.erase(branch_name);
+        rebind_branch_tree_pin(old_hash, "");
+        return 0;
+    }
+
+    // target_hash is a branch-tree root — a content-addressed blob produced
+    // by an earlier commit. Reject if unknown.
+    if (!g_rt->storage->Get(hash)) return -1;
+
+    session->branch_overrides[branch_name] = hash;
+    rebind_branch_tree_pin(old_hash, hash);
+    return 0;
+}
+
 rnt_handle_t rnt_open_handle(const char* path, const char* /*claims*/)
 {
-    if (!is_init() || !path) return nullptr;
+    if (!is_initialized() || !path) return nullptr;
     return g_rt->handler->Open(split_path(path), nullptr);
 }
 
 int rnt_close_handle(rnt_handle_t handle)
 {
-    if (!is_init() || !handle) return -1;
+    if (!is_initialized() || !handle) return -1;
     auto* h = static_cast<nt::HandlerManager::handle*>(handle);
     return g_rt->handler->Close(h) ? 0 : -1;
 }
 
-int rnt_branch_payload(rnt_handle_t handle, uint8_t** payload_out, size_t* len_out)
+int rnt_branch_target(rnt_handle_t handle, char** target_hash_out)
 {
-    if (!is_init() || !handle || !payload_out || !len_out) return -1;
+    if (!is_initialized() || !handle || !target_hash_out) return -1;
 
     auto* h = static_cast<nt::HandlerManager::handle*>(handle);
     if (!h->object || !h->object->head) return -1;
@@ -259,127 +625,231 @@ int rnt_branch_payload(rnt_handle_t handle, uint8_t** payload_out, size_t* len_o
     auto* branch = dynamic_cast<nt::ObjectManager::Branch*>(h->object->object.get());
     if (!branch) return -1;
 
-    const auto& src = branch->payload;
-    auto* buf = new uint8_t[src.size()];
-    std::memcpy(buf, src.data(), src.size());
-    *payload_out = buf;
-    *len_out     = src.size();
+    *target_hash_out = heap_str(branch->target_hash);
     return 0;
 }
 
-int rnt_branch_set_payload(rnt_handle_t handle, const uint8_t* payload, size_t len)
+int rnt_branch_advance(const char* branch_path, const char* new_hash)
 {
-    if (!is_init() || !handle) return -1;
+    if (!is_initialized() || !branch_path || !new_hash) return -1;
 
-    auto* h = static_cast<nt::HandlerManager::handle*>(handle);
-    if (!h->object || !h->object->head) return -1;
-    if (h->object->head->type->label != BRANCH) return -1;
+    auto* entry = g_rt->objects.Find(split_path(branch_path));
+    if (!entry || !entry->object || !entry->head) return -1;
+    if (entry->head->type->label != BRANCH) return -1;
 
-    auto* branch = dynamic_cast<nt::ObjectManager::Branch*>(h->object->object.get());
+    auto* branch = dynamic_cast<nt::ObjectManager::Branch*>(entry->object.get());
     if (!branch) return -1;
 
-    branch->payload.assign(payload, payload + len);
+    const std::string hash(new_hash);
+    // Non-empty branch-tree roots must already exist in the KV store. The
+    // branch tree is a content-addressed blob produced by an earlier mutation
+    // through commit_relation_update; advancing to an unknown root is treated
+    // as a caller bug.
+    if (!hash.empty() && !g_rt->storage->Get(hash)) return -1;
+
+    const std::string old_hash = branch->target_hash;
+    branch->target_hash = hash;
+    rebind_branch_tree_pin(old_hash, hash);
     return 0;
 }
 
 int rnt_register_relation(const char* path)
 {
-    if (!is_init() || !path) return -1;
+    if (!is_initialized() || !path) return -1;
     const auto parts = split_path(path);
-    if (g_rt->objects.Find(parts) != nullptr) return 0;
-    g_rt->objects.Register(
-        parts,
-        std::make_unique<nt::ObjectManager::Relation>(),
-        make_relation_type());
-    return 0;
+
+    std::vector<std::string> branch_path;
+    std::string mg_name, relation_name;
+    if (!split_branch_mg_relation(parts, branch_path, mg_name, relation_name))
+        return -1;
+
+    auto* branch = find_branch(branch_path);
+    if (!branch) return -1;
+
+    // Idempotent: if the relation is already present in the named mg's
+    // current snapshot, the call succeeds without producing a new snapshot.
+    const auto current = read_mg_relations(*branch, mg_name);
+    const bool already_present = std::any_of(
+        current.begin(), current.end(),
+        [&](const auto& e) { return e.first == relation_name; });
+    if (already_present) return 0;
+
+    // Empty merkle_root signals a relation with no tuples.
+    return commit_relation_update(*branch, mg_name, relation_name, "").empty()
+        ? -1 : 0;
 }
 
-int rnt_register_branch(const char* path, const uint8_t* payload, size_t payload_len)
+int rnt_register_branch(const char* path, const char* target_hash)
 {
-    if (!is_init() || !path) return -1;
+    if (!is_initialized() || !path) return -1;
     const auto parts = split_path(path);
     if (g_rt->objects.Find(parts) != nullptr) return 0;
 
-    auto branch  = std::make_unique<nt::ObjectManager::Branch>();
-    branch->name = parts.empty() ? "" : parts.back();
-    if (payload && payload_len > 0)
-        branch->payload.assign(payload, payload + payload_len);
+    const std::string hash = target_hash ? target_hash : "";
+    // Non-empty target_hash is a branch-tree root produced by an earlier
+    // commit; it must already exist as a KV blob.
+    if (!hash.empty() && !g_rt->storage->Get(hash)) return -1;
+
+    auto branch         = std::make_unique<nt::ObjectManager::Branch>();
+    branch->name        = parts.empty() ? "" : parts.back();
+    branch->target_hash = hash;
 
     g_rt->objects.Register(parts, std::move(branch), make_branch_type());
+    rebind_branch_tree_pin("", hash);
     return 0;
 }
 
-// Looks up the Relation object for a path; returns nullptr when not found or wrong type.
-static nt::ObjectManager::Relation* find_relation(const std::vector<std::string>& parts)
+int rnt_list_relations(const char* branch_mg_path, char** out)
 {
-    auto* entry = g_rt->objects.Find(parts);
-    if (!entry || !entry->object) return nullptr;
-    return dynamic_cast<nt::ObjectManager::Relation*>(entry->object.get());
+    if (!is_initialized() || !branch_mg_path || !out) return -1;
+    const auto parts = split_path(branch_mg_path);
+    if (parts.size() != 5) return -1;
+    if (parts[0] != "system" || parts[1] != "branches"
+     || parts[3] != "multigroups") return -1;
+
+    auto* branch = find_branch({ "system", "branches", parts[2] });
+    if (!branch) return -1;
+    const auto relations = read_mg_relations(*branch, parts[4]);
+    std::string result;
+    for (const auto& [name, root] : relations)
+        result += name + "\t" + root + "\n";
+    *out = heap_str(result);
+    return 0;
+}
+
+int rnt_list_branch_multigroups(const char* branch_path, char** out)
+{
+    if (!is_initialized() || !branch_path || !out) return -1;
+    auto* branch = find_branch(split_path(branch_path));
+    if (!branch) return -1;
+    const auto mgs = list_branch_tree(branch->target_hash);
+    std::string result;
+    for (const auto& [name, mg_hash] : mgs)
+        result += name + "\t" + mg_hash + "\n";
+    *out = heap_str(result);
+    return 0;
+}
+
+int rnt_list_snapshot_relations(const char* snapshot_hash, char** out)
+{
+    if (!is_initialized() || !snapshot_hash || !out) return -1;
+    const std::string hash(snapshot_hash);
+    const auto snap_path = split_path(("/system/snapshots/" + hash).c_str());
+    auto* entry = g_rt->objects.Find(snap_path);
+    if (!entry || !entry->object) return -1;
+    auto* mg = dynamic_cast<nt::ObjectManager::Multigroup*>(entry->object.get());
+    if (!mg) return -1;
+    const auto relations = nt::MultigroupCodec::List(*g_rt->storage, mg->merkle_root);
+    std::string result;
+    for (const auto& [name, root] : relations)
+        result += name + "\t" + root + "\n";
+    *out = heap_str(result);
+    return 0;
 }
 
 int rnt_link_tuple(const char* relation_path, const char* kv_attrs, char** hash_out)
 {
-    if (!is_init() || !relation_path || !kv_attrs || !hash_out) return -1;
+    if (!is_initialized() || !relation_path || !kv_attrs || !hash_out) return -1;
     const auto parts = split_path(relation_path);
 
-    auto* rel = find_relation(parts);
-    if (!rel) return -1;
+    std::vector<std::string> branch_path;
+    std::string mg_name, relation_name;
+    if (!split_branch_mg_relation(parts, branch_path, mg_name, relation_name))
+        return -1;
+
+    auto* branch = find_branch(branch_path);
+    if (!branch) return -1;
+
+    const std::string old_root = read_relation_root(*branch, mg_name, relation_name);
 
-    auto bytes      = nt::TupleCodec::Serialize(parse_kv(kv_attrs));
-    const auto hash = g_rt->storage->Put(std::move(bytes));
+    auto bytes           = nt::TupleCodec::Serialize(parse_kv(kv_attrs));
+    const auto tuple_hash = g_rt->storage->Put(std::move(bytes));
+    const auto tuple_bin  = nt::hex_to_bin(tuple_hash);
+    const std::string new_root =
+        nt::Merkle<nt::Hash32>::Insert(*g_rt->storage, old_root,
+                                        tuple_bin, tuple_bin);
 
-    rel->merkle_root = nt::Merkle::Insert(*g_rt->storage, rel->merkle_root, hash);
+    if (commit_relation_update(*branch, mg_name, relation_name, new_root).empty())
+        return -1;
 
-    *hash_out = heap_str(hash);
+    *hash_out = heap_str(tuple_hash);
     return 0;
 }
 
 int rnt_unlink_tuple(const char* relation_path, const char* tuple_hash)
 {
-    if (!is_init() || !relation_path || !tuple_hash) return -1;
-    auto* rel = find_relation(split_path(relation_path));
-    if (!rel) return -1;
-    rel->merkle_root = nt::Merkle::Remove(*g_rt->storage, rel->merkle_root, tuple_hash);
-    return 0;
+    if (!is_initialized() || !relation_path || !tuple_hash) return -1;
+    const auto parts = split_path(relation_path);
+
+    std::vector<std::string> branch_path;
+    std::string mg_name, relation_name;
+    if (!split_branch_mg_relation(parts, branch_path, mg_name, relation_name))
+        return -1;
+
+    auto* branch = find_branch(branch_path);
+    if (!branch) return -1;
+
+    const std::string old_root = read_relation_root(*branch, mg_name, relation_name);
+    const std::string new_root =
+        nt::Merkle<nt::Hash32>::Remove(*g_rt->storage, old_root,
+                                        nt::hex_to_bin(tuple_hash));
+
+    return commit_relation_update(*branch, mg_name, relation_name, new_root).empty()
+        ? -1 : 0;
 }
 
 int rnt_clear_relation(const char* relation_path)
 {
-    if (!is_init() || !relation_path) return -1;
-    auto* rel = find_relation(split_path(relation_path));
-    if (!rel) return -1;
-    rel->merkle_root.clear();
-    return 0;
+    if (!is_initialized() || !relation_path) return -1;
+    const auto parts = split_path(relation_path);
+
+    std::vector<std::string> branch_path;
+    std::string mg_name, relation_name;
+    if (!split_branch_mg_relation(parts, branch_path, mg_name, relation_name))
+        return -1;
+
+    auto* branch = find_branch(branch_path);
+    if (!branch) return -1;
+
+    return commit_relation_update(*branch, mg_name, relation_name, "").empty()
+        ? -1 : 0;
 }
 
 int rnt_relation_root(const char* relation_path, char** root_hash_out)
 {
-    if (!is_init() || !relation_path || !root_hash_out) return -1;
-    auto* rel = find_relation(split_path(relation_path));
-    if (!rel) return -1;
-    *root_hash_out = heap_str(rel->merkle_root);
-    return 0;
-}
+    if (!is_initialized() || !relation_path || !root_hash_out) return -1;
+    const auto parts = split_path(relation_path);
 
-int rnt_set_relation_root(const char* relation_path, const char* root_hash)
-{
-    if (!is_init() || !relation_path) return -1;
-    auto* rel = find_relation(split_path(relation_path));
-    if (!rel) return -1;
-    rel->merkle_root = root_hash ? root_hash : "";
+    std::vector<std::string> branch_path;
+    std::string mg_name, relation_name;
+    if (!split_branch_mg_relation(parts, branch_path, mg_name, relation_name))
+        return -1;
+
+    auto* branch = find_branch(branch_path);
+    if (!branch) return -1;
+
+    // Distinguish "relation registered but empty" (returns 0 with empty hash)
+    // from "relation not in this multigroup's snapshot" (returns -1). Without
+    // this check an unborn branch would silently report every relation empty.
+    const auto relations = read_mg_relations(*branch, mg_name);
+    auto it = std::find_if(relations.begin(), relations.end(),
+                           [&](const auto& e) { return e.first == relation_name; });
+    if (it == relations.end()) return -1;
+
+    *root_hash_out = heap_str(it->second);
     return 0;
 }
 
 rnt_cursor_t rnt_cursor_open(rnt_handle_t handle)
 {
-    if (!is_init() || !handle) return nullptr;
+    if (!is_initialized() || !handle) return nullptr;
     auto* h = static_cast<nt::HandlerManager::handle*>(handle);
 
     std::string merkle_root;
     if (h->object && h->object->head &&
         h->object->head->type->label == RELATION)
     {
-        auto* rel = find_relation(h->object->head->path);
+        auto* rel = dynamic_cast<nt::ObjectManager::Relation*>(h->object->object.get());
         if (rel) merkle_root = rel->merkle_root;
     }
 
@@ -388,7 +858,7 @@ rnt_cursor_t rnt_cursor_open(rnt_handle_t handle)
 
 int rnt_cursor_next(rnt_cursor_t cursor, char** tuple_out)
 {
-    if (!is_init() || !cursor || !tuple_out) return -1;
+    if (!is_initialized() || !cursor || !tuple_out) return -1;
     auto* c = static_cast<nt::CursorManager::cursor*>(cursor);
     nt::Tuple* t = g_rt->cursors->Next(c);
     if (!t) { *tuple_out = nullptr; return 0; }
@@ -398,7 +868,7 @@ int rnt_cursor_next(rnt_cursor_t cursor, char** tuple_out)
 
 int rnt_cursor_close(rnt_cursor_t cursor)
 {
-    if (!is_init() || !cursor) return -1;
+    if (!is_initialized() || !cursor) return -1;
     g_rt->cursors->Close(static_cast<nt::CursorManager::cursor*>(cursor));
     return 0;
 }
@@ -412,17 +882,24 @@ void rnt_free_bytes(uint8_t* p) { delete[] p; }
 
 rnt_plan_t rnt_plan_scan(const char* relation_path)
 {
-    if (!is_init() || !relation_path) return nullptr;
+    if (!is_initialized() || !relation_path) return nullptr;
 
     // Open a handle to the stored relation through the full manager pipeline.
+    // Open() runs NamespaceReferenceManager::Resolve internally, so the handle
+    // lands on the resolved /system/snapshots/<hash>/relations/<n> entry when
+    // the caller passed a branch-relative path.
     const auto parts = split_path(relation_path);
     auto* h = g_rt->handler->Open(parts, nullptr);
     if (!h) return nullptr;
 
-    // Seed the cursor with the current Merkle root from the ObjectManager.
+    // Read the Merkle root straight from the handle's object — Resolve has
+    // already pointed it at the snapshot-bound Relation entry.
     std::string merkle_root;
-    auto* rel = find_relation(parts);
-    if (rel) merkle_root = rel->merkle_root;
+    if (h->object && h->object->object)
+    {
+        auto* rel = dynamic_cast<nt::ObjectManager::Relation*>(h->object->object.get());
+        if (rel) merkle_root = rel->merkle_root;
+    }
 
     auto* c = g_rt->cursors->Open(h, merkle_root);
     if (!c)
@@ -507,14 +984,14 @@ void rnt_plan_free(rnt_plan_t plan)
 
 rnt_cursor_t rnt_vm_execute_plan(rnt_plan_t plan)
 {
-    if (!is_init() || !plan) return nullptr;
+    if (!is_initialized() || !plan) return nullptr;
     // VmCursor constructor transfers ownership from PlanWrapper and deletes it.
     return new VmCursor(*g_rt->cursors, static_cast<PlanWrapper*>(plan));
 }
 
 int rnt_vm_cursor_next(rnt_cursor_t vm_cursor, char** tuple_out)
 {
-    if (!is_init() || !vm_cursor || !tuple_out) return -1;
+    if (!is_initialized() || !vm_cursor || !tuple_out) return -1;
     auto* vc = static_cast<VmCursor*>(vm_cursor);
     nt::Tuple* t = vc->vm.Next(vc->root);
     if (!t) { *tuple_out = nullptr; return 0; }
@@ -524,7 +1001,7 @@ int rnt_vm_cursor_next(rnt_cursor_t vm_cursor, char** tuple_out)
 
 int rnt_vm_cursor_close(rnt_cursor_t vm_cursor)
 {
-    if (!is_init() || !vm_cursor) return -1;
+    if (!is_initialized() || !vm_cursor) return -1;
     // VmCursor destructor closes all cursors and handles.
     delete static_cast<VmCursor*>(vm_cursor);
     return 0;
diff --git a/src/RelationalNT.cpp b/src/RelationalNT.cpp
index acf7ecb..b1302ee 100644
--- a/src/RelationalNT.cpp
+++ b/src/RelationalNT.cpp
@@ -6,6 +6,7 @@
 #include "IdentityManager.h"
 #include "LifecycleManager.h"
 #include "Merkle.h"
+#include "NamespaceReferenceManager.h"
 #include "ObjectManager.h"
 #include "PermissionsManager.h"
 #include "SqliteBackend.h"
@@ -51,8 +52,10 @@ int main()
     auto store = [&](std::vector<nt::Attribute> attrs) {
         auto bytes = nt::TupleCodec::Serialize(attrs);
         auto hash  = backend.Put(std::move(bytes));
+        auto hash_bin = nt::hex_to_bin(hash);
         villager_rel->merkle_root =
-            nt::Merkle::Insert(backend, villager_rel->merkle_root, hash);
+            nt::Merkle<nt::Hash32>::Insert(backend, villager_rel->merkle_root,
+                                            hash_bin, hash_bin);
     };
     store({ { "name", "Blathers" }, { "profession", "Museum Curator" } });
     store({ { "name", "Rover"   }, { "profession", "Traveller" } });
@@ -62,8 +65,9 @@ int main()
     // --- Open a handle on villager through the full manager pipeline ---
     nt::PermissionsManager permissions;
     nt::IdentityManager identities;
-    nt::LifecycleManager lifecycles;
-    nt::HandlerManager handler(objects, permissions, identities, lifecycles);
+    nt::LifecycleManager lifecycles(objects, backend);
+    nt::NamespaceReferenceManager references(objects, backend);
+    nt::HandlerManager handler(objects, permissions, identities, lifecycles, references);
 
     int connection = 1;  // dummy connection context
     nt::HandlerManager::handle* handle = handler.Open(path, &connection);