Ephemeral relationsmaster-old

4 files changed, 373 insertions, 5 deletions

diff --git a/src/CursorManager.cpp b/src/CursorManager.cpp
index 0d71b1c..23f826f 100644
--- a/src/CursorManager.cpp
+++ b/src/CursorManager.cpp
@@ -61,17 +61,18 @@ namespace nt {
         };
 
         if (label == EPHEMERAL_RELATION) {
-            // Cursor is created exhausted — the JOIN operator writes args and
-            // 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.
+            // Cursor opens ready for a standalone scan (no bound args, offset
+            // zero) so a named ephemeral can be drained like a stored relation.
+            // A JOIN probe overwrites this state anyway: the operator writes
+            // args and resets the cursor 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->is_ephemeral = true;
             c->generator = etype->generator;
-            c->exhausted = true;
             pin_snapshot(c);
             return c;
         }
diff --git a/src/EphemeralRelation.cpp b/src/EphemeralRelation.cpp
new file mode 100644
index 0000000..5cb2dd7
--- /dev/null
+++ b/src/EphemeralRelation.cpp
@@ -0,0 +1,196 @@
+#include "EphemeralRelation.h"
+
+#include "Types.h"
+
+#include <picosha2.h>
+
+#include <algorithm>
+#include <memory>
+
+namespace nt::Ephemeral {
+    namespace {
+        // Object-type descriptor for an EPHEMERAL_RELATION. Ephemeral relations
+        // are not durable (disposable=true) and never contend (readers compose
+        // generators independently). The generator and cardinality live on the
+        // descriptor; the cursor layer copies the generator out at Open time.
+        std::unique_ptr<ObjectManager::object_type>
+        make_type(ObjectManager::ephemeral_object_type::Cardinality cardinality,
+                  ObjectManager::ephemeral_object_type::Generator generator) {
+            auto t = std::make_unique<ObjectManager::ephemeral_object_type>();
+            t->label = EPHEMERAL_RELATION;
+            t->disposable = true;
+            t->methods = {OPEN, CLOSE};
+            t->exclusive = false;
+            t->cardinality = cardinality;
+            t->generator = std::move(generator);
+            return t;
+        }
+
+        // Reads the merkle_root off a registry entry whose object is a stored or
+        // ephemeral relation. Empty string for anything else (or missing root).
+        std::string entry_root(ObjectManager::registry* entry) {
+            if (entry == nullptr || entry->object == nullptr)
+                return {};
+            if (auto* r = dynamic_cast<ObjectManager::Relation*>(entry->object.get()))
+                return r->merkle_root;
+            if (auto* e = dynamic_cast<ObjectManager::EphemeralRelation*>(entry->object.get()))
+                return e->merkle_root;
+            return {};
+        }
+    } // namespace
+
+    std::vector<std::string> SplitPath(const std::string& path) {
+        std::vector<std::string> parts;
+        std::string seg;
+        for (char c : path) {
+            if (c == '/') {
+                if (!seg.empty())
+                    parts.push_back(seg);
+                seg.clear();
+            } else {
+                seg.push_back(c);
+            }
+        }
+        if (!seg.empty())
+            parts.push_back(seg);
+        return parts;
+    }
+
+    std::string ComposeRoot(const std::string& generator_identity,
+                            const std::vector<std::pair<std::string, std::string>>& schema,
+                            const std::vector<std::string>& base_roots) {
+        // Sort schema and base roots so input order does not change the digest.
+        std::vector<std::pair<std::string, std::string>> sorted_schema = schema;
+        std::sort(sorted_schema.begin(), sorted_schema.end());
+        std::vector<std::string> sorted_roots = base_roots;
+        std::sort(sorted_roots.begin(), sorted_roots.end());
+
+        std::string buf;
+        buf += generator_identity;
+        buf.push_back('\0');
+        // NUL-separate each field so distinct (name, type) splits cannot
+        // serialize to the same bytes, then a lone NUL to close the schema.
+        for (const auto& [name, type] : sorted_schema) {
+            buf += name;
+            buf.push_back('\0');
+            buf += type;
+            buf.push_back('\0');
+        }
+        buf.push_back('\0');
+        for (const auto& root : sorted_roots) {
+            buf += root;
+            buf.push_back('\n');
+        }
+        return picosha2::hash256_hex_string(buf.begin(), buf.end());
+    }
+
+    ObjectManager::registry* Register(
+        ObjectManager& objects, LifecycleManager& lifecycles, const std::string& session_hash,
+        Tier tier, const std::string& name,
+        ObjectManager::ephemeral_object_type::Generator generator,
+        ObjectManager::ephemeral_object_type::Cardinality cardinality,
+        const std::string& generator_identity,
+        const std::vector<std::pair<std::string, std::string>>& schema,
+        const std::vector<std::string>& dependencies) {
+        if (session_hash.empty())
+            return nullptr;
+        if (tier == Tier::Named && name.empty())
+            return nullptr;
+
+        // Resolve dependencies up front: read their current roots for the
+        // identity hash and keep the entries to pin after registration. An
+        // unresolvable dependency fails the registration; skipping it would
+        // store a path that was never pinned, and CascadeEphemeral would later
+        // Unpin a hold it never took.
+        std::vector<ObjectManager::registry*> dep_entries;
+        std::vector<std::string> base_roots;
+        dep_entries.reserve(dependencies.size());
+        base_roots.reserve(dependencies.size());
+        for (const auto& dep : dependencies) {
+            auto* entry = objects.Find(SplitPath(dep));
+            if (entry == nullptr)
+                return nullptr;
+            dep_entries.push_back(entry);
+            base_roots.push_back(entry_root(entry));
+        }
+
+        const std::string root = ComposeRoot(generator_identity, schema, base_roots);
+        const std::string subdir = (tier == Tier::Named) ? "ephemeral" : "scratch";
+        const std::string leaf = (tier == Tier::Named) ? name : root;
+        const std::vector<std::string> path = {"system", "sessions", session_hash, subdir, leaf};
+
+        // Idempotent: an identical scratch result, or a live named binding,
+        // reuses the existing entry without double-pinning.
+        if (auto* existing = objects.Find(path))
+            return existing;
+
+        auto er = std::make_unique<ObjectManager::EphemeralRelation>();
+        er->merkle_root = root;
+        er->dependencies = dependencies;
+        objects.Register(path, std::move(er), make_type(cardinality, std::move(generator)));
+
+        auto* entry = objects.Find(path);
+        if (entry == nullptr)
+            return nullptr;
+
+        // Structural pins on the base relations: released by
+        // LifecycleManager::CascadeEphemeral when this entry is collected.
+        for (auto* dep : dep_entries)
+            lifecycles.Pin(dep);
+
+        // Tier policy: a named ephemeral is owned by its session and outlives
+        // its cursor; a scratch intermediate is owned by nobody and dies by
+        // counters with its cursor.
+        if (tier == Tier::Named)
+            lifecycles.Pin(entry);
+
+        return entry;
+    }
+
+    std::size_t ReleaseSession(ObjectManager& objects, LifecycleManager& lifecycles,
+                               const std::string& session_hash) {
+        const std::vector<std::string> named_prefix = {"system", "sessions", session_hash,
+                                                        "ephemeral"};
+        const std::vector<std::string> scratch_prefix = {"system", "sessions", session_hash,
+                                                          "scratch"};
+
+        auto is_child = [](const std::vector<std::string>& path,
+                           const std::vector<std::string>& prefix) {
+            if (path.size() != prefix.size() + 1)
+                return false;
+            for (std::size_t i = 0; i < prefix.size(); ++i)
+                if (path[i] != prefix[i])
+                    return false;
+            return true;
+        };
+
+        // Collect targets before mutating: Unpin / Collect splice entries out of
+        // the same list we are walking.
+        std::vector<ObjectManager::registry*> named;
+        std::vector<ObjectManager::registry*> scratch;
+        for (auto* cur = objects.entries.get(); cur != nullptr; cur = cur->next.get()) {
+            if (cur->head == nullptr)
+                continue;
+            const auto& p = cur->head->path;
+            if (is_child(p, named_prefix))
+                named.push_back(cur);
+            else if (is_child(p, scratch_prefix))
+                scratch.push_back(cur);
+        }
+
+        std::size_t released = 0;
+        // Named: drop the session-ownership pin; collects if nothing else holds it.
+        for (auto* entry : named) {
+            const uint32_t before = entry->head->reference_count;
+            lifecycles.Unpin(entry);
+            if (before > 0)
+                ++released;
+        }
+        // Scratch: never session-pinned; collect any stragglers whose cursor closed.
+        for (auto* entry : scratch)
+            if (lifecycles.Collect(entry))
+                ++released;
+
+        return released;
+    }
+} // namespace nt::Ephemeral
diff --git a/src/LifecycleManager.cpp b/src/LifecycleManager.cpp
index 4a12e09..10f3bb1 100644
--- a/src/LifecycleManager.cpp
+++ b/src/LifecycleManager.cpp
@@ -1,5 +1,6 @@
 #include "LifecycleManager.h"
 
+#include "EphemeralRelation.h"
 #include "Merkle.h"
 #include "Types.h"
 
@@ -44,6 +45,20 @@ namespace nt {
         return object != nullptr;
     }
 
+    bool LifecycleManager::Collect(ObjectManager::registry* object) {
+        if (!IsEligibleForGC(object))
+            return false;
+        // Named-lifetime types (BRANCH / SESSION) never auto-collect; TryCollect
+        // would no-op on them, so return false rather than report a collection.
+        if (object->head->type != nullptr) {
+            const auto label = object->head->type->label;
+            if (label == BRANCH || label == SESSION)
+                return false;
+        }
+        TryCollect(object);
+        return true;
+    }
+
     void LifecycleManager::TryCollect(ObjectManager::registry* object) {
         if (!IsEligibleForGC(object))
             return;
@@ -66,6 +81,8 @@ namespace nt {
             CascadeMultigroup(object);
         if (label == BRANCH_TREE)
             CascadeBranchTree(object);
+        if (label == EPHEMERAL_RELATION)
+            CascadeEphemeral(object);
 
         objects_.Unregister(object->head->path);
     }
@@ -140,4 +157,27 @@ namespace nt {
         for (auto* mg : mgs)
             Unpin(mg);
     }
+
+    void LifecycleManager::CascadeEphemeral(ObjectManager::registry* ephemeral) {
+        if (ephemeral == nullptr || ephemeral->object == nullptr)
+            return;
+        auto* er = dynamic_cast<ObjectManager::EphemeralRelation*>(ephemeral->object.get());
+        if (er == nullptr)
+            return;
+
+        // Resolve each dependency path and collect before mutating: Unpin can
+        // cascade further GC, and iteration must outlive that mutation.
+        std::vector<ObjectManager::registry*> deps;
+        for (const auto& dep : er->dependencies) {
+            const auto parts = Ephemeral::SplitPath(dep);
+            if (parts.empty())
+                continue;
+            auto* entry = objects_.Find(parts);
+            if (entry != nullptr)
+                deps.push_back(entry);
+        }
+
+        for (auto* dep : deps)
+            Unpin(dep);
+    }
 } // namespace nt
diff --git a/src/RNT_C_API.cpp b/src/RNT_C_API.cpp
index a632ab9..a83ac3a 100644
--- a/src/RNT_C_API.cpp
+++ b/src/RNT_C_API.cpp
@@ -1,6 +1,7 @@
 #include "RNT_C_API.h"
 
 #include "CursorManager.h"
+#include "EphemeralRelation.h"
 #include "HandlerManager.h"
 #include "IdentityManager.h"
 #include "InMemoryBackend.h"
@@ -751,6 +752,10 @@ int rnt_session_close(const char* session_hash) {
         rebind_branch_tree_pin(hash, "");
     session->branch_overrides.clear();
 
+    // Drop the session's owned ephemeral relations (named tier-1 bindings and
+    // any scratch stragglers), cascading their base-relation pins.
+    nt::Ephemeral::ReleaseSession(g_rt->objects, *g_rt->lifecycles, session_hash);
+
     return g_rt->objects.Unregister(session_path) ? 0 : -1;
 }
 
@@ -792,6 +797,132 @@ int rnt_session_set_branch(const char* session_hash, const char* branch_name,
     return 0;
 }
 
+// ---------------------------------------------------------------------------
+// Ephemeral relations
+// ---------------------------------------------------------------------------
+
+namespace {
+    // Accumulates tuples emitted by a generator callback during one page
+    // request. Lives on the stack of the generator wrapper below; the opaque
+    // rnt_tuple_sink_t handed to the callback points at it.
+    struct TupleSink {
+        std::vector<nt::Tuple> tuples;
+    };
+
+    // Splits a newline-separated list into its non-empty lines.
+    std::vector<std::string> split_lines(const char* s) {
+        std::vector<std::string> lines;
+        if (!s)
+            return lines;
+        std::stringstream ss(s);
+        std::string line;
+        while (std::getline(ss, line))
+            if (!line.empty())
+                lines.push_back(line);
+        return lines;
+    }
+
+    // Joins registry path segments back into a slash-separated logical path.
+    std::string join_path(const std::vector<std::string>& segs) {
+        std::string out;
+        for (const auto& seg : segs) {
+            if (!out.empty())
+                out.push_back('/');
+            out += seg;
+        }
+        return out;
+    }
+} // namespace
+
+int rnt_sink_emit(rnt_tuple_sink_t sink, const char* tuple_kv) {
+    if (!sink || !tuple_kv)
+        return -1;
+    auto* s = static_cast<TupleSink*>(sink);
+    s->tuples.emplace_back(parse_kv(tuple_kv));
+    return 0;
+}
+
+int rnt_register_ephemeral_relation(const char* session_hash, int named, const char* name,
+                                    rnt_generator_fn generator, void* generator_ctx,
+                                    int cardinality, const char* generator_identity,
+                                    const char* schema_kv, const char* dependencies,
+                                    char** path_out) {
+    if (!g_rt || !session_hash || !generator || !generator_identity)
+        return -1;
+    using Card = nt::ObjectManager::ephemeral_object_type::Cardinality;
+    if (cardinality < static_cast<int>(Card::Finite) ||
+        cardinality > static_cast<int>(Card::Continuum))
+        return -1;
+
+    // The session must exist: the entry lands under its namespace and is
+    // released by rnt_session_close.
+    auto* session = g_rt->objects.Find({"system", "sessions", session_hash});
+    if (!session || !session->head || session->head->type->label != SESSION)
+        return -1;
+
+    std::vector<std::pair<std::string, std::string>> schema;
+    for (const auto& line : split_lines(schema_kv)) {
+        const auto pos = line.find('=');
+        if (pos == std::string::npos)
+            return -1;
+        schema.emplace_back(line.substr(0, pos), line.substr(pos + 1));
+    }
+
+    // Wrap the C callback in the std::function shape the cursor layer copies
+    // out at Open time. The sink gives emitted tuples a single well-defined
+    // owner: this wrapper's stack frame, for the duration of one page request.
+    auto gen = [generator, generator_ctx](const std::vector<std::string>& args,
+                                          std::size_t offset,
+                                          std::size_t limit) -> std::vector<nt::Tuple> {
+        std::string joined;
+        for (const auto& a : args) {
+            joined += a;
+            joined.push_back('\n');
+        }
+        TupleSink sink;
+        const int rc = generator(generator_ctx, joined.c_str(), offset, limit, &sink);
+        if (rc != 0)
+            return {};
+        return std::move(sink.tuples);
+    };
+
+    // Resolve each dependency through the reference manager so callers may
+    // pass branch-relative paths (e.g. system/branches/<b>/multigroups/<mg>/
+    // relations/<r>); the stored dependency list then holds the resolved
+    // snapshot paths that raw registry lookups (and the GC cascade) address.
+    std::vector<std::string> deps;
+    for (const auto& dep : split_lines(dependencies)) {
+        const auto resolved = g_rt->references->Resolve(nt::Ephemeral::SplitPath(dep));
+        if (resolved.empty())
+            return -1;
+        deps.push_back(join_path(resolved));
+    }
+
+    auto* entry = nt::Ephemeral::Register(
+        g_rt->objects, *g_rt->lifecycles, session_hash,
+        named ? nt::Ephemeral::Tier::Named : nt::Ephemeral::Tier::Scratch,
+        name ? name : "", std::move(gen), static_cast<Card>(cardinality), generator_identity,
+        schema, deps);
+    if (!entry)
+        return -1;
+
+    if (path_out)
+        *path_out = heap_str(join_path(entry->head->path));
+    return 0;
+}
+
+int rnt_drop_ephemeral_relation(const char* session_hash, const char* name) {
+    if (!g_rt || !session_hash || !name || !*name)
+        return -1;
+    auto* entry = g_rt->objects.Find({"system", "sessions", session_hash, "ephemeral", name});
+    if (!entry || !entry->head || entry->head->type->label != EPHEMERAL_RELATION)
+        return -1;
+    // Releases the session-ownership pin taken at registration; the entry is
+    // collected once no cursor or dependent ephemeral still holds it.
+    g_rt->lifecycles->Unpin(entry);
+    return 0;
+}
+
 rnt_handle_t rnt_open_handle(const char* path, const char* /*claims*/) {
     if (!path)
         return nullptr;