state.h

Go to the documentation of this file.

 
#pragma once
 
#include <neograph/api.h>
#include <neograph/graph/channel_key.h>
#include <neograph/graph/types.h>
#include <neograph/graph/run_context.h>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <vector>
#include <core/native_archive.h>
 
namespace neograph::graph {
namespace detail { class ManagedBudgetJournalAccess; }
class NativeGraphCheckpoint final {
private:
    struct History {
        std::vector<sp::Message> messages;
        json projection = json::array();
    };
    struct BudgetBinding {
        std::shared_ptr<UsageAccumulator> bank;
        std::uint64_t ceiling = 0;
        std::uint64_t original_ceiling = 0;
        bool managed = false;
        std::string owner_scope, thread_id, graph_identity;
        std::string original_thread_id, fork_source_thread_id;
    };
    std::map<std::string, History> histories_;
    std::map<std::string, ProviderLoopHistory::Entry> loops_;
    std::vector<sp::runtime::Result> outcomes_;
    std::optional<BudgetBinding> budget_;
    json projection_;
    NativeGraphCheckpoint() = default;
    friend class GraphState;
    friend class detail::ManagedBudgetJournalAccess;
};
 
class NEOGRAPH_API GraphState {
public:
    void init_channel(const std::string& name,
                      ReducerType type,
                      ReducerFn reducer,
                      const json& initial_value = json(),
                      ChannelLifecyclePolicy lifecycle = {});
 
    json get(const std::string& channel) const;
 
    template <typename T>
    T get(const ChannelKey<T>& channel) const {
        return get(channel.name()).template get<T>();
    }
 
    template <typename T>
    std::optional<T> try_get(const ChannelKey<T>& channel) const {
        if (!has_channel(channel.name())) return std::nullopt;
        return get(channel);
    }
 
    std::vector<ChatMessage> get_messages() const;
    std::vector<sp::Message> get_provider_messages(const std::string& channel = "messages") const;
    std::optional<std::vector<sp::Message>> captured_provider_messages(
        const std::string& channel = "messages") const;
    void set_native_history_archive(std::shared_ptr<sp::NativeArchive> archive);
    void copy_provider_history_from(const GraphState& source);
    void set_provider_run_history(std::shared_ptr<ProviderLoopHistory> loops,
                                  std::shared_ptr<ProviderOutcomes> outcomes);
    std::vector<sp::runtime::Result> provider_outcomes() const;
    void configure_budget_bank(std::shared_ptr<UsageAccumulator> bank, std::uint64_t ceiling,
        bool managed, std::string owner_scope, std::string thread_id, std::string graph_identity);
    std::shared_ptr<UsageAccumulator> budget_bank() const;
    std::uint64_t budget_ceiling() const;
    std::uint64_t budget_original_ceiling() const;
    std::string budget_original_thread_id() const;
    bool budget_managed() const;
    void defer_budget_authority_restore();
    void activate_budget_authority(bool observation_only = false);
    void rebind_fork_budget_thread(std::string thread_id, bool original_in_memory_custody);
    void validate_budget_context(const std::string& graph_identity, const std::string& thread_id) const;
 
    void write(const std::string& channel, const json& value);
 
    void apply_writes(const std::vector<ChannelWrite>& writes);
 
    uint64_t channel_version(const std::string& channel) const;
 
    uint64_t global_version() const;
 
    json serialize() const;
 
    void restore(const json& data);
    json serialize_runtime() const;
    json serialize_cache() const;
    void restore_runtime(const json& data);
 
    json ephemeral_checkpoint_guard() const;
 
    void restore_checkpoint(const json& data, const json& guard,
                            std::shared_ptr<const NativeGraphCheckpoint> native = {});
    std::pair<json, std::shared_ptr<const NativeGraphCheckpoint>> checkpoint_snapshot() const;
 
 
    std::vector<std::string> channel_names() const;
 
    bool has_channel(const std::string& channel) const;
 
    // v1.0 (9d): the v0.3 `run_cancel_token_` smuggling channel is gone.
    // Cancel propagation now flows exclusively through
    // `RunContext::cancel_token` on `NodeInput::ctx` (engine threads it
    // through every dispatch; multi-Send fan-out copies the context by
    // value, so the cancel token survives the isolated-state boundary
    // without a serialize/restore hop).
 
private:
    std::map<std::string, Channel> channels_;
    uint64_t global_version_ = 0;
    using ProviderHistory = NativeGraphCheckpoint::History;
    std::map<std::string, ProviderHistory> provider_histories_;
    std::shared_ptr<sp::NativeArchive> native_history_archive_;
    std::shared_ptr<ProviderLoopHistory> provider_loops_;
    std::shared_ptr<ProviderOutcomes> provider_outcomes_;
    std::optional<NativeGraphCheckpoint::BudgetBinding> provider_budget_;
    bool defer_budget_authority_ = false;
    std::optional<UsageAccumulator::AuthoritySnapshot> deferred_budget_authority_;
    std::shared_ptr<UsageAccumulator> deferred_budget_bank_;
    void admit_budget_binding(NativeGraphCheckpoint::BudgetBinding saved,
                              std::optional<UsageAccumulator::AuthoritySnapshot> authority);
    json managed_budget_projection_locked() const;
    void update_provider_history_locked(const std::string& channel, const json& before, const json& incoming,
        const json& after, const std::shared_ptr<const std::vector<sp::Message>>& native);
    void save_provider_history_locked(json& snapshot, bool durable = true, bool include_budget = true) const;
    json serialize_runtime_locked(bool include_budget) const;
    void restore_provider_history_locked(const json& snapshot, bool trusted_runtime_copy = false);
    mutable std::shared_mutex mutex_;
};
 
} // namespace neograph::graph