Class: FuncRunner::Application

Inherits:
Object
  • Object
show all
Defined in:
lib/func_runner/application.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Application

Returns a new instance of Application.



7
8
9
10
11
12
13
14
15
 
# File 'lib/func_runner/application.rb', line 7

def initialize
  @function_registry = {}
  @is_running = false
  @logger = FuncRunner.config.logger
  @config = FuncRunner.config
  if @config.api_key.nil?
    raise "Missing Func Runner API key. Set FUNCRUNNER_API_KEY environment variable or use config.api_key="
  end
end

Instance Attribute Details

#config ⇒ Object

Returns the value of attribute config.



5
6
7
 
# File 'lib/func_runner/application.rb', line 5

def config
  @config
end

#function_registry ⇒ Object

Returns the value of attribute function_registry.



5
6
7
 
# File 'lib/func_runner/application.rb', line 5

def function_registry
  @function_registry
end

#is_running ⇒ Object

Returns the value of attribute is_running.



5
6
7
 
# File 'lib/func_runner/application.rb', line 5

def is_running
  @is_running
end

#logger ⇒ Object

Returns the value of attribute logger.



5
6
7
 
# File 'lib/func_runner/application.rb', line 5

def logger
  @logger
end

Instance Method Details

#configure_assistant ⇒ Object 



129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
 
# File 'lib/func_runner/application.rb', line 129

def configure_assistant
  logger.info("Updating assistant functions...", assistant_id: config.assistant_id)

  url = "https://proxy.funcrunner.com/v1/assistants/#{config.assistant_id}"
  response = make_request(:get, url)

  unless response&.status == 200
    logger.error("Failed to fetch assistant", url: url, error: response&.body)
    return
  end

  assistant = JSON.parse(response.body)

  # Filter and update existing tools
  updated_tools = assistant["tools"].select { |tool| %w[code_interpreter file_search].any? { |t| tool.include?(t) } }

  # Generate function specs and append them to the assistant's tools
  function_specs = function_registry.values.map { |func| generate_function_spec(func) }
  assistant["tools"] = updated_tools + function_specs

  # Remove read-only attributes to avoid errors in the update request
  %w[id object created_at].each { |attr| assistant.delete(attr) }
  assistant.delete("description") if assistant["description"].nil?

  # Update assistant with new tools
  update_response = make_request(:post, url, nil, assistant.to_json)
  if update_response&.status == 200
    logger.info("Successfully updated assistant functions.", assistant_id: config.assistant_id)
  else
    logger.error("Failed to update assistant", url: url, status_code: update_response&.status, error: update_response&.body)
  end
end

#delete_message(message_id, correlation_id) ⇒ Object 



119
120
121
122
123
124
125
126
127
 
# File 'lib/func_runner/application.rb', line 119

def delete_message(message_id, correlation_id)
  logger.info("Deleting message from the queue", message_id: message_id, correlation_id: correlation_id)
  response = make_request(:delete, "https://queue.funcrunner.com/messages/#{message_id}")
  if response&.status == 200
    logger.info("Successfully deleted queue message", message_id: message_id, correlation_id: correlation_id)
  else
    logger.error("Failed to delete message from the queue", message_id: message_id, correlation_id: correlation_id)
  end
end

#dequeue_message ⇒ Object 



95
96
97
98
99
100
101
102
103
104
 
# File 'lib/func_runner/application.rb', line 95

def dequeue_message
  response = make_request(:get, "https://queue.funcrunner.com/messages")
  return unless response&.status == 200

  messages = JSON.parse(response.body)
  unless messages.empty?
    logger.info("Message dequeued", message_id: messages[0]["id"])
    Message.new(messages[0])
  end
end

#execute_function(fe, message) ⇒ Object 



63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/func_runner/application.rb', line 63

def execute_function(fe, message)
  unless @function_registry.key?(fe.name)
    logger.error("Function not found in registry", function_name: fe.name, available_functions: @function_registry.keys, message: message.to_h, correlation_id: message.correlation_id)
    raise NotImplementedError, "Function '#{fe.name}' not found in registry."
  end

  function = @function_registry[fe.name]
  logger.info("Executing function", function_name: fe.name, arguments: fe.arguments, correlation_id: message.correlation_id)
  function.call(**fe.arguments)
rescue ArgumentError => e
  logger.error("Argument error when calling function", function_name: fe.name, error: e.message, message: message.to_h, correlation_id: message.correlation_id)
  raise "Argument error when calling '#{fe.name}': #{e.message}"
end

#extract_tool_calls(run) ⇒ Object 



77
78
79
80
 
# File 'lib/func_runner/application.rb', line 77

def extract_tool_calls(run)
  tool_calls = run.dig("required_action", "submit_tool_outputs", "tool_calls") || []
  tool_calls.is_a?(Array) ? tool_calls : []
end

#fetch_run_data(message) ⇒ Object 



24
25
26
27
28
29
30
31
32
 
# File 'lib/func_runner/application.rb', line 24

def fetch_run_data(message)
  url = "https://proxy.funcrunner.com/v1/threads/#{message.thread_id}/runs/#{message.run_id}"
  logger.debug("Fetching run data from URL: #{url}", message_id: message.id, run_id: message.run_id, correlation_id: message.correlation_id)
  response = make_request(:get, url, message)
  return JSON.parse(response.body) if response&.status == 200

  logger.error("Failed to fetch run data", url: url)
  nil
end

#function(name, &definition) ⇒ Object 



17
18
19
20
21
22
 
# File 'lib/func_runner/application.rb', line 17

def function(name, &definition)
  function_definition = FunctionDefinition.new(name)
  function_definition.instance_eval(&definition)
  @function_registry[name] = function_definition.to_h
  logger.info("Registered function '#{name}' with params #{function_definition.param_types}")
end

#make_request(method, url, message = nil, data = nil) ⇒ Object 



34
35
36
37
38
39
40
 
# File 'lib/func_runner/application.rb', line 34

def make_request(method, url, message = nil, data = nil)
  headers = {"Authorization" => "Bearer #{@config.api_key}"}
  Faraday.send(method, url, data, headers)
rescue Faraday::Error => e
  logger.error("HTTP request failed", method: method, url: url, error: e.message, correlation_id: message&.correlation_id || "N/A")
  nil
end

#process_queue_message(message) ⇒ Object 



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/func_runner/application.rb', line 42

def process_queue_message(message)
  logger.info("Processing queue message", message_id: message.id, correlation_id: message.correlation_id)
  run = fetch_run_data(message)
  return if run.nil?

  tool_calls = extract_tool_calls(run)
  function_executions = process_tool_calls(tool_calls)

  result = RunResult.new(run_id: message.run_id, thread_id: message.thread_id, tool_outputs: [])
  function_executions.each do |fe|
    output = execute_function(fe, message)
    if output.is_a?(String) || output.nil?
      result.tool_outputs << {tool_call_id: fe.tool_call_id, output: output}
    else
      logger.error("#{fe.name} returned non-string value. OpenAI expects functions to return a string.")
    end
  end

  result
end

#process_tool_calls(tool_calls) ⇒ Object 



82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/func_runner/application.rb', line 82

def process_tool_calls(tool_calls)
  tool_calls.map do |tool_call|
    name = tool_call.dig("function", "name")
    logger.info("Processing function call request", function_name: name, tool_call_id: tool_call["id"])
    arguments = begin
      JSON.parse(tool_call.dig("function", "arguments"))
    rescue
      {}
    end
    FunctionExecution.new(name: name, arguments: arguments, tool_call_id: tool_call["id"])
  end
end

#run ⇒ Object 



162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
 
# File 'lib/func_runner/application.rb', line 162

def run
  logger.info("Starting Func Runner application...")
  logger.info("Available functions: #{function_registry.keys.join(", ")}") unless function_registry.empty?
  logger.info("No functions registered.") if function_registry.empty?
  @is_running = true

  configure_assistant if config.auto_update && config.assistant_id

  begin
    while @is_running
      message = dequeue_message
      run_result = process_queue_message(message) if message
      submit_function_results(run_result) if run_result
      delete_message(message.id, message.correlation_id) if run_result
      sleep(config.polling_interval)
    end
  rescue Interrupt
    logger.info("Shutting down gracefully...")
    @is_running = false
  end
end

#stop ⇒ Object 



184
185
186
187
 
# File 'lib/func_runner/application.rb', line 184

def stop
  @is_running = false
  logger.info("Func Runner application stopped")
end

#submit_function_results(result) ⇒ Object 



106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/func_runner/application.rb', line 106

def submit_function_results(result)
  logger.info("Submitting function results", run_id: result.run_id, correlation_id: result.thread_id)
  url = "https://proxy.funcrunner.com/v1/threads/#{result.thread_id}/runs/#{result.run_id}/submit_tool_outputs"
  response = make_request(:post, url, nil, result.dump_submission_response)
  if response&.status == 200
    logger.info("Successfully submitted function results", run_id: result.run_id, correlation_id: result.thread_id)
    true
  else
    logger.error("Failed to submit function results", run_id: result.run_id, correlation_id: result.thread_id)
    false
  end
end