runtime.h

Go to the documentation of this file.

 
#pragma once
 
#include <neograph/api.h>
#include <neograph/graph/checkpoint.h>
#include <neograph/graph/configuration_failure.h>
#include <neograph/graph/store.h>
#include <neograph/host_admission.h>
#include <neograph/hook_runtime.h>
#include <neograph/program/blocking.h>
#include <neograph/program/catalog.h>
#include <neograph/program/core_tool_grant.h>
#include <neograph/program/core_provider_call.h>
#include <neograph/program/fork.h>
#include <neograph/program/graph_migration.h>
#include <neograph/program/handle.h>
#include <neograph/program/module.h>
#include <neograph/program/pending.h>
#include <neograph/program/replay.h>
#include <neograph/program/task_graph_fragment.h>
#include <neograph/program/invocation.h>
#include <neograph/program/transition_store.h>
 
#include <asio/any_io_executor.hpp>
#include <asio/awaitable.hpp>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <memory>
#include <optional>
#include <stdexcept>
#include <string>
#include <string_view>
#include <vector>
 
namespace neograph::program {
struct ProgramInvocation {
    json                              input;
    RunBudget                         budget;
    std::string                       trace_id;
    std::shared_ptr<ProgramEventSink> events;
    std::string                       requested_run_id;
    std::string                       parent_run_id;
    std::uint32_t                     child_depth = 0;
    std::optional<RunInvocation>      canonical_request;
};
 
struct ProgramResume {
    json                              value;
    std::string                       trace_id;
    std::shared_ptr<ProgramEventSink> events;
    std::string                       pending_id;
};
 
struct ProgramEffectResolution {
    std::string                       pending_id;
    ProgramEffectReconciliation       resolution = ProgramEffectReconciliation::Unknown;
    std::optional<json>                result;
    std::string                       trace_id;
    std::shared_ptr<ProgramEventSink> events;
};
 
struct ProgramGraphMigrationTarget {
    std::string                       target_program_version_id;
    std::string                       requested_run_id;
    std::shared_ptr<ProgramEventSink> events;
    std::optional<GraphSemanticMigrationAdapter> semantic_adapter;
};
 
struct ProgramHostAdmissionContext {
    std::string_view owner_scope;
    std::string_view program_version_id;
    std::string_view run_id;
    std::string_view operation_id;
    std::uint64_t    attempt = 0;
    RunBudget        granted_budget;
    std::uint32_t    child_depth = 0;
};
 
using ProgramHostAdmissionResolver =
    std::function<HostAdmissionRequest(const ProgramHostAdmissionContext&)>;
 
struct ProgramRuntimeRecoveryState {
    std::string                              owner_scope;
    std::string                              run_id;
    std::vector<ProgramContextPublication>   context_publications;
    std::vector<HookOutboxEntry>              hook_outbox_entries;
    std::vector<HookOutboxEntry>              inherited_hook_outbox_entries;
    std::optional<ProgramRuntimeStateTransferReceipt> transfer_receipt;
};
 
using ProgramRuntimeRecoveryHandler =
    std::function<void(const ProgramRuntimeRecoveryState&)>;
 
using ProgramChildSynthesisGrantResolver = std::function<std::optional<ProgramChildSynthesisGrant>(
    std::string_view owner_scope, std::string_view parent_run_id, std::string_view grant_id)>;
 
struct ProgramChildQuotaConfig {
    std::uint64_t max_active_children = 0;
    std::uint64_t max_pending_spawn_requests = 0;
    std::uint64_t max_active_children_per_owner = 0;
    std::uint64_t max_pending_spawn_requests_per_owner = 0;
};
 
struct ProgramExecutorIdentity {
    bool host_pumped = false;
    std::function<bool()> running_in_this_thread;
};
 
 struct RuntimeConfig {
     std::shared_ptr<ProgramCatalog>         catalog;
     std::shared_ptr<graph::CheckpointStore> checkpoints;
     std::shared_ptr<graph::Store>           state_store;
     std::shared_ptr<ProgramTransitionStore> transitions;
     std::size_t                              scheduler_threads = 1;
     ProgramChildBindingResolver             child_binding_resolver;
     std::shared_ptr<TaskGraphFragmentStore> task_graph_fragments;
     TaskGraphExpansionPolicyResolver        task_graph_policy_resolver;
     std::shared_ptr<HostAdmissionController> host_admission;
     ProgramHostAdmissionResolver             host_admission_resolver;
      ProgramChildQuotaConfig                  child_quota;
      std::shared_ptr<HookRuntime>              hook_runtime;
      ProgramRuntimeRecoveryHandler             runtime_recovery_handler;
      std::shared_ptr<ProgramSynthesisGateway>  child_synthesis_gateway;
      ProgramChildSynthesisGrantResolver        child_synthesis_grant_resolver;
      std::function<void(ProgramHandle, ProgramHandoff)> checkpoint_handler;
      bool recover_existing_child_commands = false;
      ProgramCoreToolGrantResolver core_tool_grant_resolver;
      ProgramCoreToolGrantAwaitableResolver core_tool_grant_awaitable_resolver;
      ProgramCoreProviderCallResolver core_provider_call_resolver;
      bool require_core_provider_call_broker = false;
      bool core_provider_calls_journaled = false;
      std::shared_ptr<sp::NativeArchive> native_history_archive;
      bool fork_may_resend_provider_calls = false;
      asio::any_io_executor                     executor;
      ProgramExecutorIdentity                   executor_identity;
  };
struct ProgramActiveRun {
    ProgramActivation activation;
    ProgramHandle handle;
};
 
class NEOGRAPH_PROGRAM_API ProgramRuntime {
public:
    explicit ProgramRuntime(RuntimeConfig config);
    ProgramRuntime(ProgramRuntime&& other) noexcept;
    ProgramRuntime& operator=(ProgramRuntime&& other) noexcept;
    ProgramRuntime(const ProgramRuntime&)            = delete;
    ProgramRuntime& operator=(const ProgramRuntime&) = delete;
    ~ProgramRuntime();
 
    ProgramHandle start(RunInvocation invocation);
    ProgramActiveRun start_active(RunInvocation invocation,
                                  std::shared_ptr<ProgramEventSink> events = {});
    ProgramHandle start(RunInvocation invocation, std::shared_ptr<ProgramEventSink> events);
 
    ProgramHandle replay_recorded(std::string_view source_run_id,
                                  RunInvocation invocation,
                                  RecordedBindingSet recorded,
                                  std::shared_ptr<ProgramEventSink> events = {});
 
    ProgramHandle start(std::string_view      owner_scope,
                        const ProgramVersion& version,
                        ProgramInvocation     invocation);
 
    ProgramHandle start_child(std::string_view         owner_scope,
                              const ProgramHandle&     parent,
                              const ModuleLinkReceipt& link,
                              const ProgramVersion&    version,
                              ProgramInvocation        invocation);
    ProgramSynthesisReservation reserve_synthesis(std::string_view                owner_scope,
                                                  const ProgramHandle&            source,
                                                  ProgramHandoff&                 handoff,
                                                  const ProgramSynthesisProposal& proposal,
                                                  std::string_view expected_lineage_head);
    ProgramChildSynthesisRecord prepare_child_synthesis(std::string_view                owner_scope,
                                                        const ProgramHandle&            parent,
                                                        const ProgramHandoff&           handoff,
                                                        const ProgramSynthesisProposal& proposal,
                                                        const ProgramChildSynthesisGrant& grant,
                                                        std::string binding_name);
    ProgramChildSynthesisRecord recover_child_synthesis(std::string_view owner_scope,
                                                        std::string_view parent_run_id,
                                                        std::string_view proposal_id);
    ProgramHandle reconnect(std::string_view owner_scope, std::string_view run_id);
    std::vector<ProgramHandle> recover_children(std::string_view owner_scope,
                                                std::string_view parent_run_id);
    ProgramHandle resume(std::string_view owner_scope,
                         std::string_view run_id,
                         ProgramResume   resume);
    ProgramHandle reconcile(std::string_view          owner_scope,
                            std::string_view          run_id,
                            ProgramEffectResolution  resolution);
    ProgramHandle fork(ExactProgramCheckpointReference source,
                       RunInvocation                 invocation,
                       ProgramResume                 resume,
                       std::shared_ptr<ProgramEventSink> events = {});
    ProgramHandle fork(std::string_view                  owner_scope,
                       ExactProgramCheckpointReference   source,
                       const ProgramVersion&             target,
                       ProgramInvocation                 invocation,
                       ProgramResume                     resume);
    ProgramHandle replace(ExactProgramHandoffReference source,
                           RunInvocation                invocation,
                           std::shared_ptr<ProgramEventSink> events = {});
    ProgramHandle replace(ProgramHandoff&&             source,
                          RunInvocation                 invocation,
                          std::shared_ptr<ProgramEventSink> events = {});
    ProgramHandle replace(std::string_view                owner_scope,
                           ExactProgramHandoffReference     source,
                           const ProgramVersion&            target,
                           ProgramInvocation                invocation);
    ProgramHandle replace(std::string_view      owner_scope,
                           ProgramHandoff&&       source,
                           const ProgramVersion&  target,
                           ProgramInvocation      invocation);
    ProgramHandle migrate_graph(const ProgramHandle&          source,
                                ProgramGraphMigrationTarget target);
 
    asio::awaitable<ProgramResult> run_async(std::string owner_scope,
                                             ProgramVersion version,
                                             ProgramInvocation invocation);
    asio::awaitable<ProgramResult> resume_async(std::string owner_scope,
                                                std::string run_id,
                                                ProgramResume resume);
    ProgramResult run(std::string_view      owner_scope,
                      const ProgramVersion& version,
                      ProgramInvocation     invocation);
 
    asio::awaitable<ProgramHandle> start_async(RunInvocation invocation,
                                               std::shared_ptr<ProgramEventSink> events = {});
    asio::awaitable<ProgramActiveRun> start_active_async(RunInvocation invocation,
                                                         std::shared_ptr<ProgramEventSink> events = {});
    asio::awaitable<ProgramHandle> start_async(std::string      owner_scope,
                                               ProgramVersion   version,
                                               ProgramInvocation invocation);
    asio::awaitable<ProgramHandle> replay_recorded_async(std::string                       source_run_id,
                                                         RunInvocation                     invocation,
                                                         RecordedBindingSet                recorded,
                                                         std::shared_ptr<ProgramEventSink> events = {});
    asio::awaitable<ProgramHandle> start_child_async(std::string         owner_scope,
                                                     ProgramHandle       parent,
                                                     ModuleLinkReceipt   link,
                                                     ProgramVersion      version,
                                                     ProgramInvocation   invocation);
    asio::awaitable<ProgramSynthesisReservation> reserve_synthesis_async(
        std::string owner_scope, ProgramHandle source, ProgramHandoff& handoff,
        ProgramSynthesisProposal proposal, std::string expected_lineage_head);
    asio::awaitable<ProgramChildSynthesisRecord> prepare_child_synthesis_async(
        std::string owner_scope, ProgramHandle parent, const ProgramHandoff& handoff,
        ProgramSynthesisProposal proposal, ProgramChildSynthesisGrant grant, std::string binding_name);
    asio::awaitable<ProgramChildSynthesisRecord> recover_child_synthesis_async(
        std::string owner_scope, std::string parent_run_id, std::string proposal_id);
    asio::awaitable<ProgramHandle> reconnect_async(std::string owner_scope, std::string run_id);
    asio::awaitable<std::vector<ProgramHandle>> recover_children_async(std::string owner_scope,
                                                                        std::string parent_run_id);
    asio::awaitable<ProgramHandle> resume_handle_async(std::string owner_scope,
                                                       std::string run_id,
                                                       ProgramResume resume);
    asio::awaitable<ProgramHandle> reconcile_async(std::string owner_scope,
                                                   std::string run_id,
                                                   ProgramEffectResolution resolution);
    asio::awaitable<ProgramHandle> fork_async(ExactProgramCheckpointReference source,
                                              RunInvocation                 invocation,
                                              ProgramResume                 resume,
                                              std::shared_ptr<ProgramEventSink> events = {});
    asio::awaitable<ProgramHandle> fork_async(std::string                     owner_scope,
                                              ExactProgramCheckpointReference source,
                                              ProgramVersion                  target,
                                              ProgramInvocation               invocation,
                                              ProgramResume                   resume);
    asio::awaitable<ProgramHandle> replace_async(ExactProgramHandoffReference source,
                                                 RunInvocation                invocation,
                                                 std::shared_ptr<ProgramEventSink> events = {});
    asio::awaitable<ProgramHandle> replace_async(ProgramHandoff                 source,
                                                 RunInvocation                  invocation,
                                                 std::shared_ptr<ProgramEventSink> events = {});
    asio::awaitable<ProgramHandle> replace_async(std::string                    owner_scope,
                                                 ExactProgramHandoffReference   source,
                                                 ProgramVersion                 target,
                                                 ProgramInvocation              invocation);
    asio::awaitable<ProgramHandle> replace_async(std::string                    owner_scope,
                                                 ProgramHandoff                 source,
                                                 ProgramVersion                 target,
                                                 ProgramInvocation              invocation);
    asio::awaitable<ProgramHandle> migrate_graph_async(ProgramHandle              source,
                                                       ProgramGraphMigrationTarget target);
 
private:
    struct Impl;
    struct Borrowed {};
    ProgramRuntime(std::shared_ptr<Impl> impl, Borrowed) noexcept;
    ProgramRuntime borrowed() const noexcept;
 
    // The coroutine body of each operation (#557). The awaitable form runs it on the runtime's executor; the
    // synchronous form runs the same body to completion on the calling thread. They take their arguments by value.
    asio::awaitable<ProgramHandle> migrate_graph_core(ProgramHandle source, ProgramGraphMigrationTarget target);
    asio::awaitable<ProgramActiveRun> start_active_core(RunInvocation invocation,
                                                        std::shared_ptr<ProgramEventSink> events);
    asio::awaitable<ProgramHandle> start_core(RunInvocation invocation, std::shared_ptr<ProgramEventSink> events);
    asio::awaitable<ProgramHandle> start_core(std::string owner_scope, ProgramVersion version,
                                              ProgramInvocation invocation);
    asio::awaitable<ProgramHandle> start_resolved_core(std::string owner_scope, ProgramVersion version,
                                                       ProgramInvocation invocation, bool active_selection = false);
    asio::awaitable<ProgramHandle> start_child_core(std::string owner_scope, ProgramHandle parent,
                                                    ModuleLinkReceipt link, ProgramVersion version,
                                                    ProgramInvocation invocation);
    asio::awaitable<ProgramHandle> replay_recorded_core(std::string source_run_id, RunInvocation invocation,
                                                        RecordedBindingSet recorded,
                                                        std::shared_ptr<ProgramEventSink> events);
    asio::awaitable<ProgramHandle> fork_core(ExactProgramCheckpointReference source, RunInvocation invocation,
                                             ProgramResume resume, std::shared_ptr<ProgramEventSink> events);
    asio::awaitable<ProgramHandle> fork_core(std::string owner_scope, ExactProgramCheckpointReference source,
                                             ProgramVersion target, ProgramInvocation invocation,
                                             ProgramResume resume);
    asio::awaitable<ProgramHandle> replace_core(ExactProgramHandoffReference source, RunInvocation invocation,
                                                std::shared_ptr<ProgramEventSink> events);
    asio::awaitable<ProgramHandle> replace_core(ProgramHandoff* source, RunInvocation invocation,
                                                std::shared_ptr<ProgramEventSink> events);
    asio::awaitable<ProgramHandle> replace_core(std::string owner_scope, ExactProgramHandoffReference source,
                                                ProgramVersion target, ProgramInvocation invocation);
    asio::awaitable<ProgramHandle> replace_core(std::string owner_scope, ProgramHandoff* source,
                                                ProgramVersion target, ProgramInvocation invocation);
    asio::awaitable<std::vector<ProgramHandle>> recover_children_core(std::string owner_scope,
                                                                       std::string parent_run_id);
    asio::awaitable<ProgramSynthesisReservation> reserve_synthesis_core(
        std::string owner_scope, ProgramHandle source, ProgramHandoff* handoff, ProgramSynthesisProposal proposal,
        std::string expected_lineage_head);
    asio::awaitable<ProgramChildSynthesisRecord> prepare_child_synthesis_core(
        std::string owner_scope, ProgramHandle parent, const ProgramHandoff* handoff,
        ProgramSynthesisProposal proposal, ProgramChildSynthesisGrant grant, std::string binding_name);
    asio::awaitable<ProgramChildSynthesisRecord> recover_child_synthesis_core(
        std::string owner_scope, std::string parent_run_id, std::string proposal_id);
    asio::awaitable<ProgramHandle> reconnect_core(std::string owner_scope, std::string run_id,
                                                  bool transition_lock_held = false);
    asio::awaitable<ProgramHandle> resume_core(std::string owner_scope, std::string run_id, ProgramResume resume);
    asio::awaitable<ProgramHandle> reconcile_core(std::string owner_scope, std::string run_id,
                                                  ProgramEffectResolution resolution);
 
    std::shared_ptr<Impl> impl_;
    bool                  borrowed_ = false;
};
 
}  // namespace neograph::program