Class: RcrewAI::Rails::Observation::Collector

Inherits:
Object
  • Object
show all
Defined in:
lib/rcrewai/rails/observation/collector.rb

Overview

Translates the flat RCrewAI event stream into a span tree.

This is the only component that knows the rcrewai event vocabulary. If that vocabulary changes, nothing outside this class moves.

Instance Method Summary collapse

Constructor Details

#initialize(execution:, writer: nil, trace_id: nil) ⇒ Collector

Returns a new instance of Collector.



15
16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/rcrewai/rails/observation/collector.rb', line 15

def initialize(execution:, writer: nil, trace_id: nil)
  @execution = execution
  @trace_id = trace_id || SecureRandom.uuid
  @stack = SpanStack.new
  @writer = writer || Writer.new(
    mode: config.observation_flush_mode,
    flush_every: config.observation_flush_every,
    on_span_change: method(:broadcast_span)
  )
  @agent_spans = {}
  @root_span_id = nil
  @text_buffers = Hash.new { |h, k| h[k] = +"" }
end

Instance Method Details

#call(event) ⇒ Object

The sink handed to crew.execute(stream:).



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/rcrewai/rails/observation/collector.rb', line 30

def call(event)
  return unless config.observation_enabled

  case event
  when RCrewAI::Events::IterationStart then on_iteration_start(event)
  when RCrewAI::Events::IterationEnd   then on_iteration_end(event)
  when RCrewAI::Events::ToolCallStart  then on_tool_start(event)
  when RCrewAI::Events::ToolCallResult then on_tool_result(event)
  when RCrewAI::Events::ToolCallError  then on_tool_error(event)
  when RCrewAI::Events::Usage          then on_usage(event)
  when RCrewAI::Events::TextDelta      then on_text_delta(event)
  when RCrewAI::Events::TextDone       then on_text_done(event)
  when RCrewAI::Events::Thinking       then on_thinking(event)
  when RCrewAI::Events::Error          then on_error(event)
  end
rescue StandardError => e
  warn_failure(e)
end

#finish! ⇒ Object

Closes anything still open. Called when the run ends, so a crash mid-span does not leave the tree permanently "running".



88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/rcrewai/rails/observation/collector.rb', line 88

def finish!
  @stack.open_span_ids.compact.each { |id| close_span(id, status: "error") }
  @agent_spans.each_value { |id| close_span(id, status: "error") }
  @agent_spans.clear
  # Close the root last: its children must be closed first so the
  # waterfall shows the run finishing after everything inside it.
  finish_crew_span(status: "error")
  # Deltas accumulate per agent and are normally freed by TextDone.
  # A stream that aborts mid-generation never sends one, so drop
  # anything left rather than retaining the whole generated text.
  @text_buffers.clear
  @writer.flush!
end

#finish_agent_span(agent_name:, status: "ok") ⇒ Object 



73
74
75
76
 
# File 'lib/rcrewai/rails/observation/collector.rb', line 73

def finish_agent_span(agent_name:, status: "ok")
  id = @agent_spans.delete(SpanStack.key_for(agent_name))
  close_span(id, status: status) if id
end

#finish_crew_span(status: "ok") ⇒ Object 



55
56
57
58
59
 
# File 'lib/rcrewai/rails/observation/collector.rb', line 55

def finish_crew_span(status: "ok")
  id = @root_span_id
  @root_span_id = nil
  close_span(id, status: status) if id
end

#finish_open_agent_spans(status: "ok") ⇒ Object

Closes every agent span opened lazily from the event stream. The caller uses this on the success path; finish! closes whatever is left as errored.



81
82
83
84
 
# File 'lib/rcrewai/rails/observation/collector.rb', line 81

def finish_open_agent_spans(status: "ok")
  @agent_spans.each_value { |id| close_span(id, status: status) }
  @agent_spans.clear
end

#start_agent_span(agent_name:, parent_span_id: nil) ⇒ Object

Agent and task spans have no corresponding events, so the engine opens them explicitly around its own dispatch.



63
64
65
66
67
68
69
70
71
 
# File 'lib/rcrewai/rails/observation/collector.rb', line 63

def start_agent_span(agent_name:, parent_span_id: nil)
  key = SpanStack.key_for(agent_name)
  id = open_span(
    kind: "agent", name: agent_name.to_s,
    parent_span_id: parent_span_id || @root_span_id, agent: agent_name
  )
  @agent_spans[key] = id
  id
end

#start_crew_span(crew_name:) ⇒ Object

Opens the run's root span. rcrewai emits no crew-level event, so the engine opens this explicitly around its own dispatch.



51
52
53
 
# File 'lib/rcrewai/rails/observation/collector.rb', line 51

def start_crew_span(crew_name:)
  @root_span_id = open_span(kind: "crew", name: crew_name.to_s)
end