Module: ClaudeSwarm::SessionCostCalculator

Extended by:
SessionCostCalculator
Included in:
SessionCostCalculator
Defined in:
lib/claude_swarm/session_cost_calculator.rb

Constant Summary collapse

MODEL_PRICING =

Model pricing in dollars per million tokens

{
  opus: {
    input: 15.0,
    output: 75.0,
    cache_write: 18.75,
    cache_read: 1.50,
  },
  sonnet: {
    input: 3.0,
    output: 15.0,
    cache_write: 3.75,
    cache_read: 0.30,
  },
  haiku: {
    input: 0.80,
    output: 4.0,
    cache_write: 1.0,
    cache_read: 0.08,
  },
}.freeze

Instance Method Summary collapse

Instance Method Details

#calculate_simple_total(session_log_path) ⇒ Object

Calculate simple total cost (for backward compatibility)



126
127
128
 
# File 'lib/claude_swarm/session_cost_calculator.rb', line 126

def calculate_simple_total(session_log_path)
  calculate_total_cost(session_log_path)[:total_cost]
end

#calculate_token_cost(usage, model_name) ⇒ Object

Calculate cost from token usage



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/claude_swarm/session_cost_calculator.rb', line 44

def calculate_token_cost(usage, model_name)
  model_type = model_type_from_name(model_name)
  return 0.0 unless model_type && usage

  pricing = MODEL_PRICING[model_type]
  return 0.0 unless pricing

  cost = 0.0

  # Regular input tokens
  if usage["input_tokens"]
    cost += (usage["input_tokens"] / 1_000_000.0) * pricing[:input]
  end

  # Output tokens
  if usage["output_tokens"]
    cost += (usage["output_tokens"] / 1_000_000.0) * pricing[:output]
  end

  # Cache creation tokens (write)
  if usage["cache_creation_input_tokens"]
    cost += (usage["cache_creation_input_tokens"] / 1_000_000.0) * pricing[:cache_write]
  end

  # Cache read tokens
  if usage["cache_read_input_tokens"]
    cost += (usage["cache_read_input_tokens"] / 1_000_000.0) * pricing[:cache_read]
  end

  cost
end

#calculate_total_cost(session_log_path) ⇒ Object

Calculate total cost from session log file Returns a hash with:

  • total_cost: Total cost in USD (sum of cost_usd for instances, token costs for main)
  • instances_with_cost: Set of instance names that have cost data


80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/claude_swarm/session_cost_calculator.rb', line 80

def calculate_total_cost(session_log_path)
  return { total_cost: 0.0, instances_with_cost: Set.new } unless File.exist?(session_log_path)

  # Track costs per instance - simple sum of cost_usd
  instance_costs = {}
  instances_with_cost = Set.new
  main_instance_cost = 0.0

  File.foreach(session_log_path) do |line|
    data = JsonHandler.parse(line)
    next if data == line # Skip unparseable lines

    instance_name = data["instance"]
    instance_id = data["instance_id"]

    # Handle main instance token-based costs
    if instance_id == "main" && data.dig("event", "type") == "assistant"
      usage = data.dig("event", "message", "usage")
      model = data.dig("event", "message", "model")
      if usage && model
        token_cost = calculate_token_cost(usage, model)
        main_instance_cost += token_cost
        instances_with_cost << instance_name if token_cost > 0
      end
    # Handle other instances with cost_usd (non-cumulative)
    elsif instance_id != "main" && data.dig("event", "type") == "result"
      # Use cost_usd (non-cumulative) instead of total_cost_usd (cumulative)
      if (cost = data.dig("event", "cost_usd"))
        instances_with_cost << instance_name
        instance_costs[instance_name] ||= 0.0
        instance_costs[instance_name] += cost
      end
    end
  end

  # Calculate total: sum of all instance costs + main instance token costs
  other_instances_cost = instance_costs.values.sum
  total_cost = other_instances_cost + main_instance_cost

  {
    total_cost: total_cost,
    instances_with_cost: instances_with_cost,
  }
end

#model_type_from_name(model_name) ⇒ Object

Determine model type from model name



30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/claude_swarm/session_cost_calculator.rb', line 30

def model_type_from_name(model_name)
  return unless model_name

  model_name_lower = model_name.downcase
  if model_name_lower.include?("opus")
    :opus
  elsif model_name_lower.include?("sonnet")
    :sonnet
  elsif model_name_lower.include?("haiku")
    :haiku
  end
end

#parse_instance_hierarchy(session_log_path) ⇒ Object

Parse instance hierarchy with costs from session log Returns a hash of instances with their cost data and relationships



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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
 
# File 'lib/claude_swarm/session_cost_calculator.rb', line 132

def parse_instance_hierarchy(session_log_path)
  instances = {}
  # Track main instance token costs
  main_instance_costs = {}

  return instances unless File.exist?(session_log_path)

  File.foreach(session_log_path) do |line|
    data = JsonHandler.parse(line)
    next if data == line # Skip unparseable lines

    instance_name = data["instance"]
    instance_id = data["instance_id"]
    calling_instance = data["calling_instance"]

    # Initialize instance data
    instances[instance_name] ||= {
      name: instance_name,
      id: instance_id,
      cost: 0.0,
      calls: 0,
      called_by: Set.new,
      calls_to: Set.new,
      has_cost_data: false,
    }

    # Track relationships
    if calling_instance && calling_instance != instance_name
      instances[instance_name][:called_by] << calling_instance

      instances[calling_instance] ||= {
        name: calling_instance,
        id: data["calling_instance_id"],
        cost: 0.0,
        calls: 0,
        called_by: Set.new,
        calls_to: Set.new,
        has_cost_data: false,
      }
      instances[calling_instance][:calls_to] << instance_name
    end

    # Handle main instance token-based costs
    if instance_id == "main" && data.dig("event", "type") == "assistant"
      usage = data.dig("event", "message", "usage")
      model = data.dig("event", "message", "model")
      if usage && model
        token_cost = calculate_token_cost(usage, model)
        if token_cost > 0
          main_instance_costs[instance_name] ||= 0.0
          main_instance_costs[instance_name] += token_cost
          instances[instance_name][:has_cost_data] = true
          instances[instance_name][:calls] += 1
        end
      end
    # Track costs and calls for non-main instances using cost_usd
    elsif data.dig("event", "type") == "result" && instance_id != "main"
      instances[instance_name][:calls] += 1
      # Use cost_usd (non-cumulative) instead of total_cost_usd
      if (cost = data.dig("event", "cost_usd"))
        instances[instance_name][:cost] += cost
        instances[instance_name][:has_cost_data] = true
      end
    end
  end

  # Set main instance costs (replace, don't add)
  main_instance_costs.each do |name, cost|
    if instances[name]
      # For main instances, use ONLY token costs, not cumulative costs
      instances[name][:cost] = cost
    end
  end

  instances
end