Module: SwarmSDK::LogCollector

Defined in:
lib/swarm_sdk/log_collector.rb

Overview

LogCollector manages subscriber callbacks for log events with filtering support.

This module acts as an emitter implementation that forwards events to user-registered callbacks. It's designed to be set as the LogStream emitter during swarm execution.

Features

  • Filtered Subscriptions: Subscribe to specific agents, event types, or swarm IDs
  • Unsubscribe Support: Remove subscriptions by ID to prevent memory leaks
  • Error Isolation: One subscriber's error doesn't break others
  • Thread Safety: Fiber-local storage for multi-threaded environments

Thread Safety for Multi-Threaded Environments (Puma, Sidekiq)

Subscriptions are stored in Fiber-local storage (Fiber) instead of class instance variables. This ensures callbacks registered in the parent thread/fiber are accessible to child fibers created by Async reactor.

Why: In Puma/Sidekiq, class instance variables (@subscriptions) are thread-isolated and don't properly propagate to child fibers. Using Fiber-local storage ensures events emitted from within Async blocks can reach registered subscriptions.

Child fibers inherit parent fiber-local storage automatically, so events emitted from agent callbacks (on_tool_call, on_end_message, etc.) executing in child fibers can still reach the parent's registered subscriptions.

Usage

# Subscribe to all events
sub_id = LogCollector.subscribe { |event| puts event }

# Subscribe to specific agent
sub_id = LogCollector.subscribe(filter: { agent: :backend }) { |event|
puts "Backend: #{event}"
}

# Subscribe to specific event types
sub_id = LogCollector.subscribe(filter: { type: ["tool_call", "tool_result"] }) { |event|
log_tool_activity(event)
}

# Subscribe with regex matching
sub_id = LogCollector.subscribe(filter: { type: /^tool_/ }) { |event|
track_tool_usage(event)
}

# Unsubscribe when done
LogCollector.unsubscribe(sub_id)

# After execution, reset for next use
LogCollector.reset!

Defined Under Namespace

Classes: Subscription

Class Method Summary collapse

Class Method Details

.clear_subscriptions ⇒ void

This method returns an undefined value.

Clear all subscriptions

Removes all subscriptions. Useful for testing or execution cleanup.



166
167
168
 
# File 'lib/swarm_sdk/log_collector.rb', line 166

def clear_subscriptions
  subscriptions.clear
end

.emit(entry) ⇒ void

This method returns an undefined value.

Emit an event to all matching subscribers

Automatically adds a timestamp if one doesn't exist. Errors in individual subscribers are isolated - one bad subscriber won't prevent others from receiving events.

Parameters:

  • entry (Hash) —

    Log event entry



185
186
187
188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/swarm_sdk/log_collector.rb', line 185

def emit(entry)
  entry_with_timestamp = ensure_timestamp(entry)

  subscriptions.each do |subscription|
    next unless subscription.matches?(entry_with_timestamp)

    begin
      subscription.callback.call(entry_with_timestamp)
    rescue StandardError => e
      # Error isolation - don't let one subscriber break others
      RubyLLM.logger.error("SwarmSDK: Subscription #{subscription.id} error: #{e.message}")
    end
  end
end

.reset! ⇒ void

This method returns an undefined value.

Reset the collector (clears subscriptions for next execution)



203
204
205
 
# File 'lib/swarm_sdk/log_collector.rb', line 203

def reset!
  Fiber[:log_subscriptions] = []
end

.subscribe(filter: {}) {|Hash| ... } ⇒ String

Subscribe to log events with optional filtering

Registers a callback that will receive events matching the filter criteria. Returns a subscription ID that can be used to unsubscribe later.

Examples:

Subscribe to all events

LogCollector.subscribe { |event| puts event }

Subscribe to specific agent's tool calls

sub_id = LogCollector.subscribe(
  filter: { agent: :backend, type: /^tool_/ }
) do |event|
  puts "Backend used tool: #{event[:tool]}"
end

Subscribe to multiple agents

LogCollector.subscribe(
  filter: { agent: [:backend, :frontend], type: "agent_stop" }
) { |e| record_completion(e) }

Parameters:

  • filter (Hash) (defaults to: {}) —

    Filter criteria (empty = all events)

    • :agent [Symbol, String, Array, Regexp] Agent name(s) to observe
    • :type [String, Array, Regexp] Event type(s) to receive
    • :swarm_id [String] Specific swarm instance
    • Any other key matches against event fields

Yields:

  • (Hash) —

    Log event entry

Returns:

  • (String) —

    Subscription ID for unsubscribe



142
143
144
145
146
 
# File 'lib/swarm_sdk/log_collector.rb', line 142

def subscribe(filter: {}, &block)
  subscription = Subscription.new(filter: filter, &block)
  subscriptions << subscription
  subscription.id
end

.subscription_count ⇒ Integer

Get current subscription count

Returns:

  • (Integer) —

    Number of active subscriptions



173
174
175
 
# File 'lib/swarm_sdk/log_collector.rb', line 173

def subscription_count
  subscriptions.size
end

.unsubscribe(subscription_id) ⇒ Subscription?

Unsubscribe by ID

Removes a subscription to prevent memory leaks and stop receiving events.

Parameters:

  • subscription_id (String) —

    ID returned from subscribe

Returns:

  • (Subscription, nil) —

    Removed subscription or nil if not found



154
155
156
157
158
159
 
# File 'lib/swarm_sdk/log_collector.rb', line 154

def unsubscribe(subscription_id)
  index = subscriptions.find_index { |s| s.id == subscription_id }
  return unless index

  subscriptions.delete_at(index)
end