Class: Stellwerk::Evaluator

Inherits:
Object
  • Object
show all
Defined in:
lib/stellwerk/evaluator.rb

Overview

Evaluates flows using pre-compiled JSON representation for maximum performance. Avoids all database queries by working directly with the compiled JSON structure.

The compiled JSON format: { "version": "1.0", "compiled_at": "...", "entry_node_ids": [...], "nodes": { "node_id": { id, name, type, config, ... } }, "adjacency": { "node_id": [{ "to": "...", "branch": "..." }] }, "reverse_adjacency": { "node_id": ["parent_id", ...] }, "sub_flows": { "flow_id": { ... embedded sub-flow ... } } }

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(compiled_json:, params:, sub_flows: {}, metadata: {}) ⇒ Evaluator

Implementation



45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/stellwerk/evaluator.rb', line 45

def initialize(compiled_json:, params:, sub_flows: {}, metadata: {})
  @compiled = deep_symbolize_keys(compiled_json)
  @nodes = (@compiled[:nodes] || {}).transform_keys(&:to_s)
  @adjacency = @compiled[:adjacency] || {}
  @reverse_adjacency = @compiled[:reverse_adjacency] || {}
  @entry_node_ids = @compiled[:entry_node_ids] || []
  @embedded_sub_flows = @compiled[:sub_flows] || {}
  @external_sub_flows = sub_flows

  @context = seed_initial_context(metadata, params)
  @calculator = Dentaku::Calculator.new
  register_collection_functions!
  @applied_nodes = []
  @errors = []
  @merge_buffers = Hash.new { |h, k| h[k] = [] }
  @outputs = nil
end

Class Method Details

.call(compiled_json:, params:, sub_flows: {}, metadata: {}) ⇒ Object

Public API



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/stellwerk/evaluator.rb', line 26

def self.call(compiled_json:, params:, sub_flows: {}, metadata: {})
  start_time = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  result = new(compiled_json: compiled_json, params: params, sub_flows: sub_flows, metadata: metadata).evaluate
  duration_ms = ((Process.clock_gettime(Process::CLOCK_MONOTONIC) - start_time) * 1000).round

  # Track metrics for observability if Metrics is available
  if defined?(Metrics)
    status = result.errors.any? ? "error" : "success"
    Metrics.increment("stellwerk.flow.executions", tags: { status: status })
    Metrics.timing("stellwerk.flow.execution_time", duration_ms)
    Metrics.histogram("stellwerk.flow.nodes_evaluated", result.applied_nodes.size)
  end

  result
end

Instance Method Details

#evaluate ⇒ Object 



63
64
65
66
67
68
69
70
71
72
 
# File 'lib/stellwerk/evaluator.rb', line 63

def evaluate
  return build_result if @nodes.empty?

  if @entry_node_ids.empty?
    @errors << "No start nodes found in flow"
    return build_result
  end

  evaluate_graph
end