Class: RcrewAI::Rails::Observation::SpanStack

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

Overview

In-memory bookkeeping for open spans. Holds no database state.

Agents may run concurrently under AsyncExecutor, so every operation is guarded by a mutex and spans are tracked per agent.

Constant Summary collapse

UNTAGGED =

Bucket for events that arrive without an agent. rcrewai's LLM clients emit Usage/TextDelta with a nil agent and rely on ToolRunner#retag to fill it in; anything that slips through lands here rather than silently merging with a real agent's stack.

"__untagged__"

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ SpanStack

Returns a new instance of SpanStack.



17
18
19
20
21
22
 
# File 'lib/rcrewai/rails/observation/span_stack.rb', line 17

def initialize
  @mutex = Mutex.new
  @sequence = 0
  @stacks = Hash.new { |h, k| h[k] = [] }
  @calls = {}
end

Class Method Details

.key_for(agent) ⇒ Object

Normalizes an agent key. Blank/nil agents share a single reserved bucket that cannot collide with a real agent name.



26
27
28
29
 
# File 'lib/rcrewai/rails/observation/span_stack.rb', line 26

def self.key_for(agent)
  key = agent.to_s
  key.empty? ? UNTAGGED : key
end

Instance Method Details

#current(agent:) ⇒ Object 



56
57
58
 
# File 'lib/rcrewai/rails/observation/span_stack.rb', line 56

def current(agent:)
  @mutex.synchronize { @stacks.fetch(self.class.key_for(agent), nil)&.last&.fetch(:id) }
end

#next_sequence ⇒ Object 



31
32
33
 
# File 'lib/rcrewai/rails/observation/span_stack.rb', line 31

def next_sequence
  @mutex.synchronize { @sequence += 1 }
end

#open_span_ids ⇒ Object 



70
71
72
 
# File 'lib/rcrewai/rails/observation/span_stack.rb', line 70

def open_span_ids
  @mutex.synchronize { @stacks.values.flatten.map { |f| f[:id] } }
end

#pop(agent:, key:) ⇒ Object

Removes and returns the most recent span matching key for agent.

Reads use fetch rather than []: the default block auto-vivifies on read, so querying unknown agents would retain a key forever.



44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/rcrewai/rails/observation/span_stack.rb', line 44

def pop(agent:, key:)
  @mutex.synchronize do
    stack = @stacks.fetch(self.class.key_for(agent), nil)
    next nil if stack.nil?

    index = stack.rindex { |frame| frame[:key] == key }
    next nil unless index

    stack.delete_at(index)[:id]
  end
end

#push(agent:, key:, id:) ⇒ Object 



35
36
37
38
 
# File 'lib/rcrewai/rails/observation/span_stack.rb', line 35

def push(agent:, key:, id:)
  @mutex.synchronize { @stacks[self.class.key_for(agent)] << { key: key, id: id } }
  id
end

#register_call(call_id:, span_id:) ⇒ Object 



60
61
62
 
# File 'lib/rcrewai/rails/observation/span_stack.rb', line 60

def register_call(call_id:, span_id:)
  @mutex.synchronize { @calls[call_id] = span_id }
end

#resolve_call(call_id:) ⇒ Object

Resolves and forgets a call id. Returns nil if never registered, which happens when a result arrives without a matching start.



66
67
68
 
# File 'lib/rcrewai/rails/observation/span_stack.rb', line 66

def resolve_call(call_id:)
  @mutex.synchronize { @calls.delete(call_id) }
end