Extern handles for embedded runtime

7 files changed, 1048 insertions, 111 deletions

diff --git a/src/CursorManager.cpp b/src/CursorManager.cpp
index 91a0183..219ba2e 100644
--- a/src/CursorManager.cpp
+++ b/src/CursorManager.cpp
@@ -1,4 +1,5 @@
 #include "CursorManager.h"
+#include "Merkle.h"
 #include "ObjectManager.h"
 #include "TupleCodec.h"
 
@@ -8,9 +9,9 @@ namespace nt
         : backend_(backend)
     {}
 
-    void CursorManager::LoadPage(cursor* c, const std::vector<std::string>& path)
+    void CursorManager::LoadPage(cursor* c)
     {
-        auto hashes = backend_.TupleHashes(path, c->fetch_offset, PAGE_SIZE);
+        auto hashes = Merkle::Page(backend_, c->merkle_root, c->fetch_offset, PAGE_SIZE);
         c->page.clear();
         c->page_position = 0;
 
@@ -26,7 +27,8 @@ namespace nt
             c->exhausted = true;
     }
 
-    CursorManager::cursor* CursorManager::Open(HandlerManager::handle* handle)
+    CursorManager::cursor* CursorManager::Open(HandlerManager::handle* handle,
+                                                const std::string& merkle_root)
     {
         if (handle == nullptr || handle->object == nullptr)
             return nullptr;
@@ -47,14 +49,17 @@ namespace nt
             return nullptr;
 
         auto* c = new cursor();
-        c->handle = handle;
-        LoadPage(c, handle->object->head->path);
+        c->handle      = handle;
+        c->merkle_root = merkle_root;
 
-        if (c->page.empty())
+        if (merkle_root.empty())
         {
-            delete c;
-            return nullptr;
+            // Empty relation: valid but immediately exhausted.
+            c->exhausted = true;
+            return c;
         }
+
+        LoadPage(c);
         return c;
     }
 
@@ -80,7 +85,7 @@ namespace nt
             return &c->page[c->page_position++];
         }
 
-        LoadPage(c, c->handle->object->head->path);
+        LoadPage(c);
         if (c->page.empty()) return nullptr;
         return &c->page[c->page_position++];
     }
diff --git a/src/InMemoryBackend.cpp b/src/InMemoryBackend.cpp
index 4d637fe..4649da0 100644
--- a/src/InMemoryBackend.cpp
+++ b/src/InMemoryBackend.cpp
@@ -1,17 +1,9 @@
 #include "InMemoryBackend.h"
 
 #include <picosha2.h>
-#include <numeric>
 
 namespace nt
 {
-    std::string InMemoryBackend::PathKey(const std::vector<std::string>& path)
-    {
-        return std::accumulate(
-            std::next(path.begin()), path.end(), path.front(),
-            [](std::string a, const std::string& b) { return std::move(a) + "/" + b; });
-    }
-
     std::string InMemoryBackend::Put(std::vector<uint8_t> value)
     {
         const std::string hash = picosha2::hash256_hex_string(value.begin(), value.end());
@@ -22,34 +14,7 @@ namespace nt
     std::optional<std::vector<uint8_t>> InMemoryBackend::Get(const std::string& hash)
     {
         auto it = kv_.find(hash);
-        if (it == kv_.end())
-            return std::nullopt;
+        if (it == kv_.end()) return std::nullopt;
         return it->second;
     }
-
-    void InMemoryBackend::LinkTuple(const std::vector<std::string>& relation_path,
-                                    const std::string& tuple_hash)
-    {
-        auto& hashes = index_[PathKey(relation_path)];
-        for (const auto& h : hashes)
-            if (h == tuple_hash) return;
-        hashes.push_back(tuple_hash);
-    }
-
-    std::vector<std::string> InMemoryBackend::TupleHashes(
-        const std::vector<std::string>& relation_path,
-        std::size_t offset,
-        std::size_t limit)
-    {
-        auto it = index_.find(PathKey(relation_path));
-        if (it == index_.end() || offset >= it->second.size())
-            return {};
-
-        const auto& all = it->second;
-        auto begin = all.begin() + static_cast<ptrdiff_t>(offset);
-        auto end   = (offset + limit <= all.size())
-                     ? begin + static_cast<ptrdiff_t>(limit)
-                     : all.end();
-        return { begin, end };
-    }
 }
diff --git a/src/Merkle.cpp b/src/Merkle.cpp
new file mode 100644
index 0000000..56ab349
--- /dev/null
+++ b/src/Merkle.cpp
@@ -0,0 +1,488 @@
+#include "Merkle.h"
+
+#include <algorithm>
+#include <cassert>
+#include <cstring>
+#include <numeric>
+#include <stdexcept>
+
+namespace nt {
+
+// ── Wire format constants ────────────────────────────────────────────────────
+
+static constexpr uint8_t TAG_LEAF     = 0x4C; // 'L'
+static constexpr uint8_t TAG_INTERNAL = 0x49; // 'I'
+
+// ── Byte-order helpers ───────────────────────────────────────────────────────
+
+static void write_u32(std::vector<uint8_t>& buf, uint32_t v)
+{
+    buf.push_back(static_cast<uint8_t>(v >> 24));
+    buf.push_back(static_cast<uint8_t>(v >> 16));
+    buf.push_back(static_cast<uint8_t>(v >>  8));
+    buf.push_back(static_cast<uint8_t>(v      ));
+}
+
+static void write_u64(std::vector<uint8_t>& buf, uint64_t v)
+{
+    buf.push_back(static_cast<uint8_t>(v >> 56));
+    buf.push_back(static_cast<uint8_t>(v >> 48));
+    buf.push_back(static_cast<uint8_t>(v >> 40));
+    buf.push_back(static_cast<uint8_t>(v >> 32));
+    buf.push_back(static_cast<uint8_t>(v >> 24));
+    buf.push_back(static_cast<uint8_t>(v >> 16));
+    buf.push_back(static_cast<uint8_t>(v >>  8));
+    buf.push_back(static_cast<uint8_t>(v      ));
+}
+
+static uint32_t read_u32(const uint8_t* p)
+{
+    return (uint32_t(p[0]) << 24) | (uint32_t(p[1]) << 16)
+         | (uint32_t(p[2]) <<  8) |  uint32_t(p[3]);
+}
+
+static uint64_t read_u64(const uint8_t* p)
+{
+    return (uint64_t(p[0]) << 56) | (uint64_t(p[1]) << 48)
+         | (uint64_t(p[2]) << 40) | (uint64_t(p[3]) << 32)
+         | (uint64_t(p[4]) << 24) | (uint64_t(p[5]) << 16)
+         | (uint64_t(p[6]) <<  8) |  uint64_t(p[7]);
+}
+
+// ── Hex / binary conversion ──────────────────────────────────────────────────
+
+Merkle::Hash32 Merkle::hex_to_bin(const std::string& hex)
+{
+    if (hex.size() != 64)
+        throw std::invalid_argument(
+            "Merkle: hash must be exactly 64 hex characters, got " +
+            std::to_string(hex.size()));
+
+    Hash32 out{};
+    auto nibble = [](char c) -> uint8_t {
+        if (c >= '0' && c <= '9') return static_cast<uint8_t>(c - '0');
+        if (c >= 'a' && c <= 'f') return static_cast<uint8_t>(c - 'a' + 10);
+        if (c >= 'A' && c <= 'F') return static_cast<uint8_t>(c - 'A' + 10);
+        throw std::invalid_argument(std::string("Merkle: invalid hex character '") + c + "'");
+    };
+    for (size_t i = 0; i < 32; ++i)
+        out[i] = static_cast<uint8_t>((nibble(hex[2*i]) << 4) | nibble(hex[2*i+1]));
+    return out;
+}
+
+std::string Merkle::bin_to_hex(const Hash32& bin)
+{
+    static const char HEX[] = "0123456789abcdef";
+    std::string out(64, '0');
+    for (size_t i = 0; i < 32; ++i)
+    {
+        out[2*i]   = HEX[(bin[i] >> 4) & 0xF];
+        out[2*i+1] = HEX[bin[i] & 0xF];
+    }
+    return out;
+}
+
+// ── Node serialisation ───────────────────────────────────────────────────────
+
+std::vector<uint8_t> Merkle::encode_leaf(const LeafNode& n)
+{
+    std::vector<uint8_t> buf;
+    buf.reserve(1 + 4 + n.hashes.size() * 32);
+    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());
+    return buf;
+}
+
+std::vector<uint8_t> Merkle::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);
+    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());
+        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)
+{
+    uint32_t count = read_u32(bytes.data() + 1);
+    LeafNode n;
+    n.hashes.reserve(count);
+    const uint8_t* p = bytes.data() + 5;
+    for (uint32_t i = 0; i < count; ++i, p += 32)
+    {
+        Hash32 h;
+        std::memcpy(h.data(), p, 32);
+        n.hashes.push_back(h);
+    }
+    return n;
+}
+
+Merkle::InternalNode Merkle::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)
+    {
+        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);
+    }
+    return n;
+}
+
+bool Merkle::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)
+{
+    auto opt = store.Get(hash_hex);
+    if (!opt)
+        throw std::runtime_error("Merkle: node not found: " + hash_hex);
+    return std::move(*opt);
+}
+
+std::string Merkle::store_leaf(IStorageBackend& store, const LeafNode& n)
+{
+    return store.Put(encode_leaf(n));
+}
+
+std::string Merkle::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)
+{
+    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();
+    }
+    auto node = decode_internal(bytes);
+    assert(!node.entries.empty() && "Merkle invariant: internal node must never be empty");
+    return node.entries.front().min_hash;
+}
+
+// ── Routing helper ───────────────────────────────────────────────────────────
+
+size_t Merkle::route(const InternalNode& node, const Hash32& 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");
+
+    // Rightmost child whose min_hash ≤ target.
+    // Defaults to 0 (leftmost) when target < all min_hashes.
+    size_t idx = 0;
+    for (size_t i = 1; i < node.entries.size(); ++i)
+    {
+        if (node.entries[i].min_hash <= target)
+            idx = i;
+        else
+            break;
+    }
+    return idx;
+}
+
+// ── Insert ───────────────────────────────────────────────────────────────────
+
+std::string Merkle::Insert(IStorageBackend& store,
+                             const std::string& root_hex,
+                             const std::string& hash_hex)
+{
+    const Hash32 tuple_hash = hex_to_bin(hash_hex);
+
+    if (root_hex.empty())
+    {
+        LeafNode leaf;
+        leaf.hashes.push_back(tuple_hash);
+        return store_leaf(store, leaf);
+    }
+
+    auto result = insert_into(store, root_hex, tuple_hash);
+    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;
+
+    InternalNode new_root;
+
+    ChildEntry left;
+    left.leaf_count = result.leaf_count;
+    left.min_hash   = left_min;
+    left.child_hash = hex_to_bin(result.new_hash);
+    new_root.entries.push_back(left);
+
+    ChildEntry right;
+    right.leaf_count = result.split_leaf_count;
+    right.min_hash   = right_min;
+    right.child_hash = hex_to_bin(result.split_hash);
+    new_root.entries.push_back(right);
+
+    return store_internal(store, new_root);
+}
+
+Merkle::InsertResult Merkle::insert_into(IStorageBackend& store,
+                                           const std::string& node_hex,
+                                           const Hash32& tuple_hash)
+{
+    auto bytes = load_node(store, node_hex);
+
+    // ── Leaf node ───────────────────────────────────────────────────────────
+    if (is_leaf_bytes(bytes))
+    {
+        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)
+        {
+            // Already present — no-op.
+            InsertResult r;
+            r.new_hash   = node_hex;
+            r.leaf_count = leaf.hashes.size();
+            return r;
+        }
+
+        leaf.hashes.insert(pos, tuple_hash);
+
+        if (leaf.hashes.size() <= B)
+        {
+            InsertResult r;
+            r.new_hash   = store_leaf(store, leaf);
+            r.leaf_count = leaf.hashes.size();
+            return r;
+        }
+
+        // Split leaf at midpoint.
+        size_t mid = leaf.hashes.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() };
+
+        InsertResult r;
+        r.new_hash         = store_leaf(store, left_leaf);
+        r.leaf_count       = left_leaf.hashes.size();
+        r.did_split        = true;
+        r.split_leaf_count = right_leaf.hashes.size();
+        r.split_min_hash   = right_leaf.hashes.front();
+        r.split_hash       = store_leaf(store, right_leaf);
+        return r;
+    }
+
+    // ── Internal node ───────────────────────────────────────────────────────
+    auto node = decode_internal(bytes);
+
+    size_t child_idx = route(node, tuple_hash);
+    auto child_result = insert_into(store,
+                                     bin_to_hex(node.entries[child_idx].child_hash),
+                                     tuple_hash);
+
+    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).
+    if (child_idx == 0)
+        node.entries[0].min_hash = subtree_min(store, child_result.new_hash);
+
+    uint64_t total = 0;
+    for (const auto& e : node.entries) total += e.leaf_count;
+
+    if (!child_result.did_split)
+    {
+        InsertResult r;
+        r.new_hash   = store_internal(store, node);
+        r.leaf_count = total;
+        return r;
+    }
+
+    // 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.child_hash = hex_to_bin(child_result.split_hash);
+    node.entries.insert(node.entries.begin() + static_cast<ptrdiff_t>(child_idx) + 1, sibling);
+    total += child_result.split_leaf_count;
+
+    if (node.entries.size() <= B)
+    {
+        InsertResult r;
+        r.new_hash   = store_internal(store, node);
+        r.leaf_count = total;
+        return r;
+    }
+
+    // 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() };
+
+    uint64_t left_count = 0, right_count = 0;
+    for (const auto& e : left_node.entries)  left_count  += e.leaf_count;
+    for (const auto& e : right_node.entries) right_count += e.leaf_count;
+
+    InsertResult r;
+    r.new_hash         = store_internal(store, left_node);
+    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_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)
+{
+    if (root_hex.empty()) return "";
+    const Hash32 tuple_hash = hex_to_bin(hash_hex);
+    auto result = remove_from(store, root_hex, tuple_hash);
+    if (result.new_hash.empty()) return "";
+
+    // Collapse: if the root is an internal node with a single child, skip it.
+    auto bytes = load_node(store, result.new_hash);
+    if (!is_leaf_bytes(bytes))
+    {
+        auto node = decode_internal(bytes);
+        if (node.entries.size() == 1)
+            return bin_to_hex(node.entries[0].child_hash);
+    }
+    return result.new_hash;
+}
+
+Merkle::RemoveResult Merkle::remove_from(IStorageBackend& store,
+                                           const std::string& node_hex,
+                                           const Hash32& tuple_hash)
+{
+    auto bytes = load_node(store, node_hex);
+
+    // ── Leaf node ───────────────────────────────────────────────────────────
+    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())
+            return { "", 0 };
+        return { store_leaf(store, leaf), leaf.hashes.size() };
+    }
+
+    // ── Internal node ───────────────────────────────────────────────────────
+    auto node = decode_internal(bytes);
+
+    size_t child_idx = route(node, tuple_hash);
+    auto child_result = remove_from(store,
+                                     bin_to_hex(node.entries[child_idx].child_hash),
+                                     tuple_hash);
+
+    if (child_result.new_hash.empty())
+    {
+        node.entries.erase(node.entries.begin() + static_cast<ptrdiff_t>(child_idx));
+        if (node.entries.empty()) return { "", 0 };
+    }
+    else
+    {
+        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);
+    }
+
+    uint64_t total = 0;
+    for (const auto& e : node.entries) total += e.leaf_count;
+    return { store_internal(store, node), total };
+}
+
+// ── Page ─────────────────────────────────────────────────────────────────────
+
+std::vector<std::string> Merkle::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);
+    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));
+    return out;
+}
+
+void Merkle::page_from(IStorageBackend& store,
+                        const std::string& node_hex,
+                        size_t& offset,
+                        size_t limit,
+                        std::vector<Hash32>& 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();
+
+        if (offset >= available)
+        {
+            offset -= available;
+            return;
+        }
+
+        for (size_t i = offset; i < available && out.size() < limit; ++i)
+            out.push_back(leaf.hashes[i]);
+
+        offset = 0;
+        return;
+    }
+
+    auto node = decode_internal(bytes);
+
+    for (const auto& entry : node.entries)
+    {
+        if (out.size() >= limit) break;
+
+        if (offset >= entry.leaf_count)
+        {
+            offset -= entry.leaf_count;
+            continue;
+        }
+
+        page_from(store, bin_to_hex(entry.child_hash), offset, limit, out);
+    }
+}
+
+} // namespace nt
diff --git a/src/PermissionsManager.cpp b/src/PermissionsManager.cpp
index 228dd2d..ddd49eb 100644
--- a/src/PermissionsManager.cpp
+++ b/src/PermissionsManager.cpp
@@ -14,6 +14,6 @@ namespace nt
         // like user group, policy, etc.
         // Yet to decide if this should be summoned at every attempt to get ahold of an object,
         // that might produce significant overhead. How can we solve this?
-        return object != nullptr && connection_context != nullptr;
+        return object != nullptr;
     }
 }
diff --git a/src/RNT_C_API.cpp b/src/RNT_C_API.cpp
new file mode 100644
index 0000000..34bbacc
--- /dev/null
+++ b/src/RNT_C_API.cpp
@@ -0,0 +1,531 @@
+#include "RNT_C_API.h"
+
+#include "CursorManager.h"
+#include "HandlerManager.h"
+#include "IdentityManager.h"
+#include "LifecycleManager.h"
+#include "Merkle.h"
+#include "ObjectManager.h"
+#include "PermissionsManager.h"
+#include "InMemoryBackend.h"
+#include "SqliteBackend.h"
+#include "TupleCodec.h"
+#include "VM.h"
+
+#include <cstring>
+#include <memory>
+#include <mutex>
+#include <sstream>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+// ---------------------------------------------------------------------------
+// Global runtime state
+//
+// A single Runtime instance is created per process on the first successful
+// rnt_init call.  All managers share this instance; they do not hold their
+// own state.  Initialisation is protected by g_init_mutex so that concurrent
+// first-callers are safe; once g_init_done is true the mutex is never
+// contended again.
+// ---------------------------------------------------------------------------
+
+namespace
+{
+    struct Runtime
+    {
+        std::unique_ptr<nt::IStorageBackend> storage;
+        nt::ObjectManager      objects;
+        nt::PermissionsManager permissions;
+        nt::IdentityManager    identities;
+        nt::LifecycleManager   lifecycles;
+        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; }
+
+    static std::vector<std::string> split_path(const char* path)
+    {
+        std::vector<std::string> parts;
+        if (!path) return parts;
+        std::string s(path);
+        if (!s.empty() && s[0] == '/') s.erase(0, 1);
+        std::stringstream ss(s);
+        std::string part;
+        while (std::getline(ss, part, '/'))
+            if (!part.empty()) parts.push_back(part);
+        return parts;
+    }
+
+    static std::vector<nt::Attribute> parse_kv(const char* kv)
+    {
+        std::vector<nt::Attribute> attrs;
+        if (!kv) return attrs;
+        std::stringstream ss(kv);
+        std::string line;
+        while (std::getline(ss, line))
+        {
+            if (line.empty()) continue;
+            const auto pos = line.find('=');
+            if (pos == std::string::npos) continue;
+            attrs.push_back({ line.substr(0, pos), line.substr(pos + 1) });
+        }
+        return attrs;
+    }
+
+    static std::string tuple_to_kv(nt::Tuple* t)
+    {
+        std::string out;
+        t->Reset();
+        const nt::Attribute* a = t->Next();
+        while (a)
+        {
+            out += a->name + "=" + a->value + "\n";
+            a = t->Next();
+        }
+        return out;
+    }
+
+    static char* heap_str(const std::string& s)
+    {
+        char* p = new char[s.size() + 1];
+        std::memcpy(p, s.data(), s.size());
+        p[s.size()] = '\0';
+        return p;
+    }
+
+    static std::unique_ptr<nt::ObjectManager::object_type> make_relation_type()
+    {
+        auto t      = std::make_unique<nt::ObjectManager::object_type>();
+        t->label    = RELATION;
+        t->disposable = false;
+        t->methods  = { OPEN, CLOSE };
+        t->exclusive = false;
+        return t;
+    }
+
+    static std::unique_ptr<nt::ObjectManager::object_type> make_branch_type()
+    {
+        auto t      = std::make_unique<nt::ObjectManager::object_type>();
+        t->label    = BRANCH;
+        t->disposable = false;
+        t->methods  = { OPEN, CLOSE };
+        // Branches are mutable reference heads: serialize writers through
+        // LifecycleManager::Contention.
+        t->exclusive = true;
+        return t;
+    }
+
+    // -----------------------------------------------------------------------
+    // Plan builder internals
+    // -----------------------------------------------------------------------
+
+    /**
+     * Wraps an entire plan subtree.  All PlanNode allocations and all
+     * CursorManager::cursor* / HandlerManager::handle* opened by SCAN nodes
+     * are collected here so they can be released atomically when the plan is
+     * freed or transferred to a VmCursor.
+     */
+    struct PlanWrapper
+    {
+        nt::PlanNode*                                root     = nullptr;
+        std::vector<std::unique_ptr<nt::PlanNode>>   nodes;    // owns every node in subtree
+        std::vector<nt::CursorManager::cursor*>      cursors;  // one per SCAN leaf
+        std::vector<nt::HandlerManager::handle*>     handles;  // one per SCAN leaf
+    };
+
+    static void free_plan_wrapper(PlanWrapper* pw)
+    {
+        if (!pw) return;
+        for (auto* c : pw->cursors)  g_rt->cursors->Close(c);
+        for (auto* h : pw->handles)  g_rt->handler->Close(h);
+        delete pw;
+    }
+
+    /**
+     * A VM cursor owns the materialised plan tree and all resources opened
+     * during plan construction.  After rnt_vm_execute_plan the caller drives
+     * it with rnt_vm_cursor_next / rnt_vm_cursor_close.
+     */
+    struct VmCursor
+    {
+        nt::VM                                       vm;
+        nt::PlanNode*                                root;
+        std::vector<std::unique_ptr<nt::PlanNode>>   nodes;
+        std::vector<nt::CursorManager::cursor*>      cursors;
+        std::vector<nt::HandlerManager::handle*>     handles;
+
+        explicit VmCursor(nt::CursorManager& cm, PlanWrapper* pw)
+            : vm(cm), root(pw->root)
+        {
+            nodes   = std::move(pw->nodes);
+            cursors = std::move(pw->cursors);
+            handles = std::move(pw->handles);
+            delete pw;  // struct itself; resources transferred above
+        }
+
+        ~VmCursor()
+        {
+            for (auto* c : cursors) g_rt->cursors->Close(c);
+            for (auto* h : handles) g_rt->handler->Close(h);
+        }
+    };
+}
+
+// ---------------------------------------------------------------------------
+// Public C API
+//
+// Section order mirrors the NT open-handle pipeline:
+//   Runtime lifecycle → Authentication → Handle lifecycle →
+//   Object registration → Tuple storage → Cursor / VM plan builder
+// ---------------------------------------------------------------------------
+
+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>();
+
+        const std::string drv = driver ? driver : "sqlite";
+        if (drv == "memory")
+            rt->storage = std::make_unique<nt::InMemoryBackend>();
+        else
+            rt->storage = std::make_unique<nt::SqliteBackend>(
+                storage_path ? storage_path : ":memory:");
+
+        rt->handler = std::make_unique<nt::HandlerManager>(
+            rt->objects, rt->permissions, rt->identities, rt->lifecycles);
+        rt->cursors = std::make_unique<nt::CursorManager>(*rt->storage);
+        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)
+        {
+            auto branch  = std::make_unique<nt::ObjectManager::Branch>();
+            branch->name = "master";
+            g_rt->objects.Register(main_path, std::move(branch),
+                                   make_branch_type());
+        }
+
+        g_init_done = true;
+        return 0;
+    }
+    catch (...)
+    {
+        g_rt.reset();
+        return -1;
+    }
+}
+
+int rnt_firewall(const char* /*auth_method*/, char** claims_out)
+{
+    if (!is_init() || !claims_out) return -1;
+    // PermissionsManager::Firewall is stubbed; grant all claims for now.
+    *claims_out = heap_str("READ WRITE");
+    return 0;
+}
+
+rnt_handle_t rnt_open_handle(const char* path, const char* /*claims*/)
+{
+    if (!is_init() || !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;
+    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)
+{
+    if (!is_init() || !handle || !payload_out || !len_out) 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* 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();
+    return 0;
+}
+
+int rnt_branch_set_payload(rnt_handle_t handle, const uint8_t* payload, size_t len)
+{
+    if (!is_init() || !handle) 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* branch = dynamic_cast<nt::ObjectManager::Branch*>(h->object->object.get());
+    if (!branch) return -1;
+
+    branch->payload.assign(payload, payload + len);
+    return 0;
+}
+
+int rnt_register_relation(const char* path)
+{
+    if (!is_init() || !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;
+}
+
+int rnt_register_branch(const char* path, const uint8_t* payload, size_t payload_len)
+{
+    if (!is_init() || !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);
+
+    g_rt->objects.Register(parts, std::move(branch), make_branch_type());
+    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)
+{
+    auto* entry = g_rt->objects.Find(parts);
+    if (!entry || !entry->object) return nullptr;
+    return dynamic_cast<nt::ObjectManager::Relation*>(entry->object.get());
+}
+
+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;
+    const auto parts = split_path(relation_path);
+
+    auto* rel = find_relation(parts);
+    if (!rel) return -1;
+
+    auto bytes      = nt::TupleCodec::Serialize(parse_kv(kv_attrs));
+    const auto hash = g_rt->storage->Put(std::move(bytes));
+
+    rel->merkle_root = nt::Merkle::Insert(*g_rt->storage, rel->merkle_root, hash);
+
+    *hash_out = heap_str(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;
+}
+
+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;
+}
+
+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;
+}
+
+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 : "";
+    return 0;
+}
+
+rnt_cursor_t rnt_cursor_open(rnt_handle_t handle)
+{
+    if (!is_init() || !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);
+        if (rel) merkle_root = rel->merkle_root;
+    }
+
+    return g_rt->cursors->Open(h, merkle_root);
+}
+
+int rnt_cursor_next(rnt_cursor_t cursor, char** tuple_out)
+{
+    if (!is_init() || !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; }
+    *tuple_out = heap_str(tuple_to_kv(t));
+    return 1;
+}
+
+int rnt_cursor_close(rnt_cursor_t cursor)
+{
+    if (!is_init() || !cursor) return -1;
+    g_rt->cursors->Close(static_cast<nt::CursorManager::cursor*>(cursor));
+    return 0;
+}
+
+void rnt_free_string(char* s)  { delete[] s; }
+void rnt_free_bytes(uint8_t* p) { delete[] p; }
+
+// ---------------------------------------------------------------------------
+// VM plan builder
+// ---------------------------------------------------------------------------
+
+rnt_plan_t rnt_plan_scan(const char* relation_path)
+{
+    if (!is_init() || !relation_path) return nullptr;
+
+    // Open a handle to the stored relation through the full manager pipeline.
+    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.
+    std::string merkle_root;
+    auto* rel = find_relation(parts);
+    if (rel) merkle_root = rel->merkle_root;
+
+    auto* c = g_rt->cursors->Open(h, merkle_root);
+    if (!c)
+    {
+        g_rt->handler->Close(h);
+        return nullptr;
+    }
+
+    auto  node        = std::make_unique<nt::PlanNode>();
+    node->op          = nt::PlanNode::Op::SCAN;
+    node->scan_cursor = c;
+
+    auto* pw       = new PlanWrapper();
+    pw->root       = node.get();
+    pw->nodes.push_back(std::move(node));
+    pw->cursors.push_back(c);
+    pw->handles.push_back(h);
+    return pw;
+}
+
+rnt_plan_t rnt_plan_join(rnt_plan_t left, rnt_plan_t right)
+{
+    if (!left || !right)
+    {
+        free_plan_wrapper(static_cast<PlanWrapper*>(left));
+        free_plan_wrapper(static_cast<PlanWrapper*>(right));
+        return nullptr;
+    }
+
+    auto* l = static_cast<PlanWrapper*>(left);
+    auto* r = static_cast<PlanWrapper*>(right);
+
+    auto  node  = std::make_unique<nt::PlanNode>();
+    node->op    = nt::PlanNode::Op::JOIN;
+    node->left  = l->root;
+    node->right = r->root;
+
+    auto* pw  = new PlanWrapper();
+    pw->root  = node.get();
+    pw->nodes.push_back(std::move(node));
+
+    // Absorb child resources into the new wrapper.
+    for (auto& n : l->nodes)   pw->nodes.push_back(std::move(n));
+    for (auto& n : r->nodes)   pw->nodes.push_back(std::move(n));
+    for (auto* c : l->cursors) pw->cursors.push_back(c);
+    for (auto* c : r->cursors) pw->cursors.push_back(c);
+    for (auto* h : l->handles) pw->handles.push_back(h);
+    for (auto* h : r->handles) pw->handles.push_back(h);
+
+    delete l;
+    delete r;
+    return pw;
+}
+
+rnt_plan_t rnt_plan_take(rnt_plan_t source, size_t limit)
+{
+    if (!source) return nullptr;
+
+    auto* s = static_cast<PlanWrapper*>(source);
+
+    auto  node         = std::make_unique<nt::PlanNode>();
+    node->op           = nt::PlanNode::Op::TAKE;
+    node->left         = s->root;
+    node->take_limit   = limit;
+
+    auto* pw  = new PlanWrapper();
+    pw->root  = node.get();
+    pw->nodes.push_back(std::move(node));
+
+    for (auto& n : s->nodes)   pw->nodes.push_back(std::move(n));
+    for (auto* c : s->cursors) pw->cursors.push_back(c);
+    for (auto* h : s->handles) pw->handles.push_back(h);
+
+    delete s;
+    return pw;
+}
+
+void rnt_plan_free(rnt_plan_t plan)
+{
+    free_plan_wrapper(static_cast<PlanWrapper*>(plan));
+}
+
+rnt_cursor_t rnt_vm_execute_plan(rnt_plan_t plan)
+{
+    if (!is_init() || !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;
+    auto* vc = static_cast<VmCursor*>(vm_cursor);
+    nt::Tuple* t = vc->vm.Next(vc->root);
+    if (!t) { *tuple_out = nullptr; return 0; }
+    *tuple_out = heap_str(tuple_to_kv(t));
+    return 1;
+}
+
+int rnt_vm_cursor_close(rnt_cursor_t vm_cursor)
+{
+    if (!is_init() || !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 e82b8c9..acf7ecb 100644
--- a/src/RelationalNT.cpp
+++ b/src/RelationalNT.cpp
@@ -5,6 +5,7 @@
 #include "HandlerManager.h"
 #include "IdentityManager.h"
 #include "LifecycleManager.h"
+#include "Merkle.h"
 #include "ObjectManager.h"
 #include "PermissionsManager.h"
 #include "SqliteBackend.h"
@@ -40,15 +41,18 @@ int main()
     type->disposable = false;
     type->methods = { OPEN, CLOSE };
 
-    objects.Register({ "multigroups", "sakura", "relations", "villager" }, std::make_unique<nt::ObjectManager::Relation>(), std::move(type));
+    const std::vector<std::string> path = { "multigroups", "sakura", "relations", "villager" };
+    objects.Register(path, std::make_unique<nt::ObjectManager::Relation>(), std::move(type));
+    auto* villager_rel = dynamic_cast<nt::ObjectManager::Relation*>(
+        objects.Find(path)->object.get());
 
-    // --- Populate SQLite backend with three tuples ---
+    // --- Populate backend with three tuples via the Merkle tree ---
     nt::SqliteBackend backend;
-    const std::vector<std::string> path = { "multigroups", "sakura", "relations", "villager" };
     auto store = [&](std::vector<nt::Attribute> attrs) {
         auto bytes = nt::TupleCodec::Serialize(attrs);
-        auto hash  = backend.Put(bytes);
-        backend.LinkTuple(path, hash);
+        auto hash  = backend.Put(std::move(bytes));
+        villager_rel->merkle_root =
+            nt::Merkle::Insert(backend, villager_rel->merkle_root, hash);
     };
     store({ { "name", "Blathers" }, { "profession", "Museum Curator" } });
     store({ { "name", "Rover"   }, { "profession", "Traveller" } });
@@ -62,7 +66,7 @@ int main()
     nt::HandlerManager handler(objects, permissions, identities, lifecycles);
 
     int connection = 1;  // dummy connection context
-    nt::HandlerManager::handle* handle = handler.Open({ "multigroups", "sakura", "relations", "villager" }, &connection);
+    nt::HandlerManager::handle* handle = handler.Open(path, &connection);
     if (handle == nullptr)
     {
         std::cout << "Failed to open handle for villager\n";
@@ -70,7 +74,7 @@ int main()
     }
 
     // --- Open a cursor on the relation ---
-    nt::CursorManager::cursor* cursor = cursors.Open(handle);
+    nt::CursorManager::cursor* cursor = cursors.Open(handle, villager_rel->merkle_root);
     if (cursor == nullptr)
     {
         std::cout << "Failed to open cursor for villager\n";
diff --git a/src/SqliteBackend.cpp b/src/SqliteBackend.cpp
index 4741033..7e9c9d5 100644
--- a/src/SqliteBackend.cpp
+++ b/src/SqliteBackend.cpp
@@ -1,18 +1,10 @@
 #include "SqliteBackend.h"
 
 #include <picosha2.h>
-#include <numeric>
 #include <stdexcept>
 
 namespace nt
 {
-    static std::string JoinPath(const std::vector<std::string>& path)
-    {
-        return std::accumulate(
-            std::next(path.begin()), path.end(), path.front(),
-            [](std::string a, const std::string& b) { return std::move(a) + "/" + b; });
-    }
-
     SqliteBackend::SqliteBackend(const std::string& path)
     {
         if (sqlite3_open(path.c_str(), &db_) != SQLITE_OK)
@@ -22,11 +14,6 @@ namespace nt
             "CREATE TABLE IF NOT EXISTS kv ("
             "  hash  TEXT PRIMARY KEY,"
             "  value BLOB NOT NULL"
-            ");"
-            "CREATE TABLE IF NOT EXISTS relation_index ("
-            "  path       TEXT NOT NULL,"
-            "  tuple_hash TEXT NOT NULL,"
-            "  UNIQUE(path, tuple_hash)"
             ");";
 
         char* err = nullptr;
@@ -43,11 +30,6 @@ namespace nt
         sqlite3_close(db_);
     }
 
-    std::string SqliteBackend::PathKey(const std::vector<std::string>& path)
-    {
-        return JoinPath(path);
-    }
-
     std::string SqliteBackend::Put(std::vector<uint8_t> value)
     {
         const std::string hash = picosha2::hash256_hex_string(value.begin(), value.end());
@@ -57,10 +39,10 @@ namespace nt
             "INSERT OR IGNORE INTO kv (hash, value) VALUES (?, ?)",
             -1, &stmt, nullptr);
         sqlite3_bind_text(stmt, 1, hash.c_str(), -1, SQLITE_TRANSIENT);
-        sqlite3_bind_blob(stmt, 2, value.data(), static_cast<int>(value.size()), SQLITE_TRANSIENT);
+        sqlite3_bind_blob(stmt, 2, value.data(), static_cast<int>(value.size()),
+                          SQLITE_TRANSIENT);
         sqlite3_step(stmt);
         sqlite3_finalize(stmt);
-
         return hash;
     }
 
@@ -82,42 +64,4 @@ namespace nt
         sqlite3_finalize(stmt);
         return result;
     }
-
-    void SqliteBackend::LinkTuple(const std::vector<std::string>& relation_path,
-                                  const std::string& tuple_hash)
-    {
-        const std::string path_key = PathKey(relation_path);
-
-        sqlite3_stmt* stmt = nullptr;
-        sqlite3_prepare_v2(db_,
-            "INSERT OR IGNORE INTO relation_index (path, tuple_hash) VALUES (?, ?)",
-            -1, &stmt, nullptr);
-        sqlite3_bind_text(stmt, 1, path_key.c_str(), -1, SQLITE_TRANSIENT);
-        sqlite3_bind_text(stmt, 2, tuple_hash.c_str(), -1, SQLITE_TRANSIENT);
-        sqlite3_step(stmt);
-        sqlite3_finalize(stmt);
-    }
-
-    std::vector<std::string> SqliteBackend::TupleHashes(
-        const std::vector<std::string>& relation_path,
-        std::size_t offset,
-        std::size_t limit)
-    {
-        const std::string path_key = PathKey(relation_path);
-
-        sqlite3_stmt* stmt = nullptr;
-        sqlite3_prepare_v2(db_,
-            "SELECT tuple_hash FROM relation_index"
-            " WHERE path = ? ORDER BY rowid LIMIT ? OFFSET ?",
-            -1, &stmt, nullptr);
-        sqlite3_bind_text(stmt, 1, path_key.c_str(), -1, SQLITE_TRANSIENT);
-        sqlite3_bind_int64(stmt, 2, static_cast<sqlite3_int64>(limit));
-        sqlite3_bind_int64(stmt, 3, static_cast<sqlite3_int64>(offset));
-
-        std::vector<std::string> hashes;
-        while (sqlite3_step(stmt) == SQLITE_ROW)
-            hashes.emplace_back(reinterpret_cast<const char*>(sqlite3_column_text(stmt, 0)));
-        sqlite3_finalize(stmt);
-        return hashes;
-    }
 }