Kimirock/chatwoot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(monitors): raise call limits and preserve quota retry holds (#16132)

Browse Source 
Conversation monitors now allow **10,000,000 calls per UTC month** and
**500,000 calls per UTC day**, per account. Allowances count attempted
provider requests, including preview calls, and no longer depend on
estimated token lengths. Pending evaluations retain their quota holds
when new messages arrive and resume after the applicable reset without
retrying ahead of a provider cooldown.

## What changed

- Enforce both call limits using persisted daily usage under the account
lock, so concurrent workers cannot exceed the remaining allowance. Each
attempted provider batch consumes one call, even when it evaluates
multiple monitors or returns a provider error.
- Remove the daily token reservation and reconciliation, which mixed
JSON request bytes with provider input tokens. Keep request-size limits
and provider token/cost telemetry.
- Preserve daily and monthly quota holds across new conversation input.
Spread retries over up to four hours after the later of the quota reset
and the longest provider cooldown recorded across failed batches.
- Report when the current monthly allowance was reached, ignoring
timestamps from the previous 100,000-call allowance.
- Document the RubyLLM bundle-audit exceptions for the pinned Ruby 3.4.4
runtime, with reassessment required when the runtime or supported
dependency versions change.

## Quota logging

When a committed call reaches the daily or monthly account cap, emit one
structured `conversation_monitor_limit_reached` log event for that
period. It includes `account_id`, `period` (`daily` or `monthly`),
`used`, `limit`, and the UTC reset timestamp `resets_at`. Rolled-back
calls and subsequent blocked retries do not emit events. Reaching both
caps in the same call emits one event for each period.

The event uses the existing Rails log forwarding path. In New Relic,
group events by account and period with:

```sql
FROM Log
WITH aparse(message, '%"account_id":*,%') AS account_id,
     aparse(message, '%"period":"*"%') AS period
SELECT count(*)
WHERE message LIKE '%conversation_monitor_limit_reached%'
FACET account_id, period
SINCE 7 days ago
LIMIT MAX
```

## Closes

Fixes https://linear.app/chatwoot/issue/INF-117

## How to reproduce

1. Exhaust an account's daily allowance, then receive additional
conversation messages before the next UTC midnight. Previously, each
message could pull the held work into an early retry and accumulate
another job at the reset. The work now retains its saved retry time
while recording the new input.
2. Have an earlier provider batch return a long `Retry-After`, then
exhaust the daily or monthly allowance before the next batch. The saved
retry now honors both the allowance reset and the longer provider
cooldown before adding jitter.

## Rollout

Existing quota holds retain their saved retry times; the increased
allowances apply when work next reserves a call. Already-scheduled jobs
from before deployment will still run once.

---------

Co-authored-by: Sony Mathew <2040199+sony-mathew@users.noreply.github.com>
This commit is contained in:
Vishnu Narayanan
2026-10-05 15:48:37 -07:00
committed by GitHub
parent 12ce7b73cc
commit 1202b5a2da
12 changed files with 223 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.bundler-audit.yml
View File

@@ -29,3 +29,9 @@ ignore:
# Patched 2.0 releases need an ai-agents migration (currently constrained to RubyLLM 1.x).
# Remove when a supported patched RubyLLM/ai-agents combination is adopted.
- CVE-2026-67991
# Think-tag parsing and Mistral matching ReDoS are reported on Ruby 3.1.
# We pin Ruby 3.4.4, where the affected regexes are linear-time.
# Remove when a supported patched RubyLLM/ai-agents combination is adopted;
# reassess if the supported Ruby runtime changes.
- CVE-2026-67987
- CVE-2026-67989

2
.env.example
View File

@@ -23,7 +23,7 @@ FRONTEND_URL=http://0.0.0.0:3000
# CONVERSATION_MONITORS_LIMIT=20
# Allowlisted contact/conversation custom attribute keys to send as customer context
# CONVERSATION_MONITORS_ATTRIBUTES=deployment_type
# CONVERSATION_MONITORS_DAILY_TOKEN_LIMIT=20000000
# Each account can make 500,000 monitor calls per UTC day and 10,000,000 per UTC month.
# To use a dedicated URL for help center pages
# HELPCENTER_URL=http://0.0.0.0:3000

9
enterprise/app/models/conversation_monitors/daily_usage.rb
View File

@@ -11,12 +11,17 @@ class ConversationMonitors::DailyUsage < ApplicationRecord
resets_at = now.beginning_of_month.next_month
records = where(account_id: account_id, usage_date: first_day...resets_at.to_date)
.order(:usage_date).pluck(:usage_date, :calls_count, :limit_reached_at)
used = records.sum { |(_, count, _)| count }
limit = ConversationMonitors::Configuration::MONTHLY_CALL_LIMIT
used = 0
limit_reached_at = nil
records.each do |_, count, reached_at|
used += count
limit_reached_at ||= reached_at if used >= limit
end
counts = records.to_h { |date, count, _| [date, count] }
{
limit: limit, used: used, remaining: [limit - used, 0].max, limit_reached: used >= limit,
limit_reached_at: records.filter_map(&:last).min&.to_i, resets_at: resets_at.to_i,
limit_reached_at: limit_reached_at&.to_i, resets_at: resets_at.to_i,
period_start: first_day.iso8601, timezone: 'UTC',
daily: (first_day..now.to_date).map { |date| { date: date.iso8601, calls: counts.fetch(date, 0) } }
}

2
enterprise/app/models/conversation_monitors/work_item.rb
View File

@@ -72,7 +72,7 @@ class ConversationMonitors::WorkItem < ApplicationRecord
end
def schedule_next_attempt
return if error_code == 'monthly_limit' && due_at&.future?
return if %w[monthly_limit budget_limit].include?(error_code) && due_at&.future?
self.due_at = [due_at, 3.seconds.from_now].compact.min
self.error_code = nil

3
enterprise/app/services/conversation_monitors/configuration.rb
View File

@@ -3,7 +3,8 @@ class ConversationMonitors::Configuration
ENDPOINT = Captain::JevClient::ENDPOINT
THRESHOLD = 0.6
MAX_MONITORS = 20
MONTHLY_CALL_LIMIT = 100_000
MONTHLY_CALL_LIMIT = 10_000_000
DAILY_CALL_LIMIT = 500_000
MAX_CONTEXT_BYTES = 28_000
MAX_CUSTOMER_CONTEXT_BYTES = 4_000
LIVE_MESSAGE_LIMIT = 5

13
enterprise/app/services/conversation_monitors/decision_service.rb
View File

@@ -13,16 +13,12 @@ class ConversationMonitors::DecisionService
body = request_body(state, monitors)
raise CustomExceptions::MonitorEvaluationError, 'context_limit' if body.bytesize > ConversationMonitors::Configuration::MAX_REQUEST_BYTES
usage = ConversationMonitors::Usage.new(@account_id)
usage.reserve!(body.bytesize)
ConversationMonitors::Usage.new(@account_id).reserve!
started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
data = request_decision(body)
reconcile_usage(usage, body.bytesize, data.fetch('usage').fetch('input_tokens'))
instrument(data, started, monitors.size)
data
rescue Captain::JevClient::HTTPError => e
# Rejected requests do not reach the model, but still consume a monthly call credit.
reconcile_usage(usage, body.bytesize, 0) if e.status.between?(400, 499)
raise monitor_error(e)
rescue JSON::ParserError
raise CustomExceptions::MonitorEvaluationError, 'invalid_response'
@@ -43,13 +39,6 @@ class ConversationMonitors::DecisionService
private
def reconcile_usage(usage, reserved, used)
usage.reconcile!(reserved, used)
rescue Redis::BaseError, ConnectionPool::TimeoutError => e
# Keep the successful response without retrying or refunding an uncertain token adjustment.
Rails.logger.warn("Conversation monitor token reconciliation failed: account_id=#{@account_id} error=#{e.class.name}")
end
def request_body(state, monitors)
model = self.class.resolve_model(monitors.first.model)
Captain::JevClient.request_body(model: model, state: state, questions: questions(monitors))

19
enterprise/app/services/conversation_monitors/evaluator.rb
View File

@@ -2,6 +2,7 @@ class ConversationMonitors::Evaluator
LEASE_DURATION = 2.minutes
MAX_ATTEMPTS = 5
QUESTIONS_PER_REQUEST = 20
LIMIT_RETRY_JITTER = 4.hours
def initialize(work)
@work = work
@@ -163,11 +164,13 @@ class ConversationMonitors::Evaluator
end
def next_retry_error
@errors.find { |error| error.code == 'monthly_limit' } || @errors.find(&:retryable?)
@errors.find { |error| error.code == 'monthly_limit' } ||
@errors.find { |error| error.code == 'budget_limit' } ||
@errors.find(&:retryable?)
end
def update_progress(error)
if error&.code == 'monthly_limit'
if %w[monthly_limit budget_limit].include?(error&.code)
@work.full_history_revision = @work.revision
elsif @errors.empty?
@work.processed_revision = @snapshot[:revision]
@@ -175,14 +178,18 @@ class ConversationMonitors::Evaluator
end
def next_due(newer_input, error)
return error.retry_after.seconds.from_now if error&.code == 'monthly_limit'
return (retry_delay + rand(LIMIT_RETRY_JITTER.to_i)).seconds.from_now if %w[monthly_limit budget_limit].include?(error&.code)
return 3.seconds.from_now if newer_input
return unless error
return if @work.attempts + 1 >= MAX_ATTEMPTS && error.code != 'budget_limit'
return if @work.attempts + 1 >= MAX_ATTEMPTS
delay = [error.retry_after.to_i, (5 * (2**[@work.attempts, 8].min)) + rand(5)].max
delay.seconds.from_now
retry_delay.seconds.from_now
end
def retry_delay
retry_after = @errors.select(&:retryable?).map { |batch_error| batch_error.retry_after.to_i }.max.to_i
[retry_after, (5 * (2**[@work.attempts, 8].min)) + rand(5)].max
end
def newer_input?

59
enterprise/app/services/conversation_monitors/usage.rb
View File

@@ -3,8 +3,8 @@ class ConversationMonitors::Usage
@account_id = account_id
end
def reserve!(bytes)
return unless reserve_credit!(bytes)
def reserve!
return unless reserve_credit!
ConversationMonitors::Monitor.visible.where(account_id: @account_id).pluck(:id).each do |monitor_id|
ConversationMonitors::BroadcastJob.schedule(monitor_id)
@@ -15,54 +15,41 @@ class ConversationMonitors::Usage
ConversationMonitors::DailyUsage.monthly_snapshot(@account_id, now)
end
def reconcile!(reserved, used)
Redis::Alfred.with { |redis| redis.incrby("conversation_monitors:#{daily_key}", used - reserved) }
end
private
def reserve_credit!(bytes)
tokens_reserved = false
def reserve_credit!
# Serialize the monthly sum and daily increment across all of an account's workers.
# The credit commits before the provider call, which runs outside this lock.
Account.find(@account_id).with_lock do
now = Time.current.utc
@usage_date = now.to_date
usage = snapshot(now)
raise CustomExceptions::MonitorEvaluationError.new('monthly_limit', retry_after: (usage[:resets_at] - now.to_f).ceil) if usage[:limit_reached]
reserve_tokens!(bytes)
tokens_reserved = true
limit_reached = usage[:used] + 1 == usage[:limit]
if usage[:daily].last[:calls] >= ConversationMonitors::Configuration::DAILY_CALL_LIMIT
raise CustomExceptions::MonitorEvaluationError.new('budget_limit', retry_after: (now.tomorrow.beginning_of_day - now).ceil)
end
limit_reached = usage[:remaining] == 1
ConversationMonitors::DailyUsage.record_call!(@account_id, at: now, limit_reached: limit_reached)
log_reached_limits(usage, now)
limit_reached
end
rescue StandardError
reconcile!(bytes, 0) if tokens_reserved
raise
end
def reserve_tokens!(bytes)
key = "conversation_monitors:#{daily_key}"
limit = Integer(ENV.fetch('CONVERSATION_MONITORS_DAILY_TOKEN_LIMIT', 20_000_000))
Redis::Alfred.with do |redis|
loop do
reserved = redis.watch(key) do
if redis.get(key).to_i + bytes > limit
delay = 1.day.from_now.utc.beginning_of_day - Time.current
raise CustomExceptions::MonitorEvaluationError.new('budget_limit', retry_after: delay.ceil)
end
redis.multi do |transaction|
transaction.incrby(key, bytes)
transaction.expire(key, 2.days.to_i)
end
end
break if reserved
end
def log_reached_limits(usage, now)
log_limit_reached('daily', used: usage[:daily].last[:calls] + 1,
limit: ConversationMonitors::Configuration::DAILY_CALL_LIMIT, resets_at: now.tomorrow.beginning_of_day.to_i)
log_limit_reached('monthly', used: usage[:used] + 1, limit: usage[:limit], resets_at: usage[:resets_at])
end
def log_limit_reached(period, used:, limit:, resets_at:)
return unless used == limit
ActiveRecord.after_all_transactions_commit do
Rails.logger.info({
event: 'conversation_monitor_limit_reached', account_id: @account_id,
period: period, used: used, limit: limit, resets_at: resets_at
}.to_json)
end
end
def daily_key
"account:#{@account_id}:tokens:#{(@usage_date || Time.current.utc.to_date).strftime('%Y%m%d')}"
end
end

14
spec/enterprise/controllers/api/v1/accounts/monitors_controller_spec.rb
View File

@@ -110,13 +110,17 @@ RSpec.describe 'Monitors API', type: :request do
end
it 'returns shared account usage on the list and every monitor without hiding historical reports' do
travel_to(Time.utc(2026, 10, 31, 12))
other_monitor = create(:conversation_monitor, account: account)
now = Time.current.utc
ConversationMonitors::DailyUsage.create!(account: account, usage_date: now.to_date, calls_count: 100_000, limit_reached_at: now)
(1..19).each do |days_ago|
ConversationMonitors::DailyUsage.create!(account: account, usage_date: now.to_date - days_ago, calls_count: 500_000)
end
ConversationMonitors::DailyUsage.create!(account: account, usage_date: now.to_date, calls_count: 500_000, limit_reached_at: now)
get base, headers: headers
usage = response.parsed_body.dig('meta', 'usage')
expect(usage).to include('limit' => 100_000, 'used' => 100_000, 'remaining' => 0, 'limit_reached' => true,
expect(usage).to include('limit' => 10_000_000, 'used' => 10_000_000, 'remaining' => 0, 'limit_reached' => true,
'limit_reached_at' => now.to_i, 'resets_at' => now.beginning_of_month.next_month.to_i)
[monitor, other_monitor].each do |record|
get "#{base}/#{record.id}/timeseries", headers: headers, params: query
@@ -128,8 +132,12 @@ RSpec.describe 'Monitors API', type: :request do
end
it 'applies the same allowance to preview and does not send requests after exhaustion' do
travel_to(Time.utc(2026, 10, 31, 12))
message
ConversationMonitors::DailyUsage.create!(account: account, usage_date: Time.current.utc.to_date, calls_count: 100_000,
(1..19).each do |days_ago|
ConversationMonitors::DailyUsage.create!(account: account, usage_date: Time.current.utc.to_date - days_ago, calls_count: 500_000)
end
ConversationMonitors::DailyUsage.create!(account: account, usage_date: Time.current.utc.to_date, calls_count: 500_000,
limit_reached_at: Time.current)
ConversationMonitors::PreviewJob.write(preview_key, { status: 'pending', condition: 'refund' })
ConversationMonitors::PreviewJob.perform_now(account.id, admin.id, 'sample')

52
spec/enterprise/services/conversation_monitors/decision_service_spec.rb
View File

@@ -13,7 +13,6 @@ RSpec.describe ConversationMonitors::DecisionService do
end
end
let(:endpoint) { "#{ConversationMonitors::Configuration::ENDPOINT}/v1/systemone" }
let(:token_key) { "conversation_monitors:account:#{account.id}:tokens:#{Time.current.utc.strftime('%Y%m%d')}" }
let(:response_body) do
{ id: 'gen-test', model: 'typesafe/jev-1.13-20260917', provider: 'TypeSafe',
answers: { monitor.id.to_s => { type: 'noul', noul: 0.9 } }, usage: { input_tokens: 100, output_tokens: 10, cost: 0.00003 } }
@@ -24,55 +23,18 @@ RSpec.describe ConversationMonitors::DecisionService do
.to_return(status: 200, body: response_body.to_json)
end
after { Redis::Alfred.delete(token_key) }
context 'when reconciling a successful response' do
let(:usage) { ConversationMonitors::Usage.new(account.id) }
before do
allow(ConversationMonitors::Usage).to receive(:new).with(account.id).and_return(usage)
allow(Rails.logger).to receive(:warn)
end
[Redis::CannotConnectError, Redis::TimeoutError, ConnectionPool::TimeoutError].each do |error_class|
it "preserves the response and reservation after #{error_class.name}" do
allow(usage).to receive(:reconcile!).and_raise(error_class, 'test outage')
reserved_bytes = nil
stub_request(:post, endpoint).to_return do |request|
reserved_bytes = request.body.bytesize
{ status: 200, body: response_body.to_json }
end
expect(evaluate).to eq(response_body.deep_stringify_keys)
expect(Redis::Alfred.get(token_key).to_i).to eq(reserved_bytes)
expect(usage.snapshot[:used]).to eq(1)
expect(WebMock).to have_requested(:post, endpoint).once
expect(Rails.logger).to have_received(:warn).with(
"Conversation monitor token reconciliation failed: account_id=#{account.id} error=#{error_class.name}"
)
end
end
it 'does not swallow unrelated accounting errors' do
allow(usage).to receive(:reconcile!).and_raise(ArgumentError, 'invalid adjustment')
expect { evaluate }.to raise_error(ArgumentError, 'invalid adjustment')
end
end
it 'does not call the provider when Redis cannot reserve the token budget' do
it 'counts a successful provider call without Redis token accounting' do
allow(Redis::Alfred).to receive(:with).and_raise(Redis::CannotConnectError, 'test outage')
expect { evaluate }.to raise_error(Redis::CannotConnectError)
expect(ConversationMonitors::Usage.new(account.id).snapshot[:used]).to eq(0)
expect(WebMock).not_to have_requested(:post, endpoint)
expect(evaluate).to eq(response_body.deep_stringify_keys)
expect(ConversationMonitors::Usage.new(account.id).snapshot[:used]).to eq(1)
expect(WebMock).to have_requested(:post, endpoint).once
end
it 'releases the reservation and never calls the provider when credit persistence fails' do
it 'does not charge the account or call the provider when credit persistence fails' do
allow(ConversationMonitors::DailyUsage).to receive(:record_call!).and_raise(ActiveRecord::StatementInvalid)
expect { evaluate }.to(raise_error { |error| expect(error.class.name).to eq('ActiveRecord::StatementInvalid') })
expect(Redis::Alfred.get(token_key).to_i).to eq(0)
expect(ConversationMonitors::Usage.new(account.id).snapshot[:used]).to eq(0)
expect(WebMock).not_to have_requested(:post, endpoint)
end
@@ -82,17 +44,15 @@ RSpec.describe ConversationMonitors::DecisionService do
expect { evaluate }.to raise_error(CustomExceptions::MonitorEvaluationError, 'provider_unavailable')
expect(ConversationMonitors::Usage.new(account.id).snapshot[:used]).to eq(1)
expect(Redis::Alfred.get(token_key).to_i).to be_positive
expect(WebMock).to have_requested(:post, endpoint).once
end
{ 401 => 'credentials_invalid', 429 => 'provider_busy' }.each do |status, code|
it "releases token capacity after a #{status} rejection while retaining the call credit" do
it "retains the call credit after a #{status} rejection" do
stub_request(:post, endpoint).to_return(status: status)
expect { evaluate }.to raise_error(CustomExceptions::MonitorEvaluationError, code)
expect(ConversationMonitors::Usage.new(account.id).snapshot[:used]).to eq(1)
expect(Redis::Alfred.get(token_key).to_i).to eq(0)
end
end

118
spec/enterprise/services/conversation_monitors/evaluator_spec.rb
View File

@@ -52,10 +52,7 @@ RSpec.describe ConversationMonitors::Evaluator do
expect(monitor.automation_deliveries.count).to eq(1)
end
it 'persists successful answers without another provider call when Redis reconciliation fails' do
redis = Redis::Alfred.with { |connection| connection }
allow(redis).to receive(:incrby).and_raise(Redis::CannotConnectError, 'test outage')
it 'persists successful answers without another provider call' do
evaluate.call
expect(monitor.evaluations.sole).to have_attributes(status: 'matched', score: 0.95)
@@ -159,21 +156,26 @@ RSpec.describe ConversationMonitors::Evaluator do
end
it 'holds exhausted accounts until next month and preserves full history across new input' do
travel_to(Time.utc(2026, 10, 31, 12))
now = Time.current.utc
ConversationMonitors::DailyUsage.create!(account: account, usage_date: now.to_date, calls_count: 100_000, limit_reached_at: now)
reset = now.beginning_of_month.next_month
(1..19).each do |days_ago|
ConversationMonitors::DailyUsage.create!(account: account, usage_date: now.to_date - days_ago, calls_count: 500_000)
end
ConversationMonitors::DailyUsage.create!(account: account, usage_date: now.to_date, calls_count: 500_000, limit_reached_at: now)
work.update!(attempts: 10)
evaluate.call
expect(WebMock).not_to have_requested(:post, endpoint)
expect(work.reload).to have_attributes(error_code: 'monthly_limit', due_at: be_within(2.seconds).of(now.beginning_of_month.next_month),
expect(work.reload).to have_attributes(error_code: 'monthly_limit', due_at: be_between(reset, reset + 4.hours),
full_history_revision: be > work.processed_revision)
expect(monitor.evaluations.sole.error_code).to eq('monthly_limit')
due_at = work.due_at
create(:message, account: account, conversation: conversation, content: 'Another message during the quota hold')
expect(work.reload).to have_attributes(due_at: due_at, error_code: 'monthly_limit')
travel_to(now.beginning_of_month.next_month + 3.seconds) do
travel_to(due_at + 1.second) do
evaluate.call
expect(WebMock).to have_requested(:post, endpoint).once
expect(monitor.evaluations.sole.status).to eq('matched')
@@ -181,18 +183,106 @@ RSpec.describe ConversationMonitors::Evaluator do
end
end
it 'prioritizes the monthly hold when an earlier batch failed using the final credit' do
it 'preserves the monthly hold through an earlier batch provider cooldown after the reset' do
travel_to(Time.utc(2026, 10, 31, 23, 55))
now = Time.current.utc
ConversationMonitors::DailyUsage.create!(account: account, usage_date: now.to_date, calls_count: 99_999)
create_list(:conversation_monitor, 19, account: account, created_at: 1.minute.ago)
stub_request(:post, endpoint).to_return(status: 429)
reset = now.beginning_of_month.next_month
(1..19).each do |days_ago|
ConversationMonitors::DailyUsage.create!(account: account, usage_date: now.to_date - days_ago, calls_count: 500_000)
end
ConversationMonitors::DailyUsage.create!(account: account, usage_date: now.to_date, calls_count: 499_999)
create_list(:conversation_monitor, 18, account: account, condition: '退款' * 250, created_at: 1.minute.ago)
create(:message, account: account, conversation: conversation, content: 'x' * 30_000)
stub_request(:post, endpoint).to_return(status: 429, headers: { 'Retry-After' => '7200' })
evaluator = described_class.new(work.reload)
allow(evaluator).to receive(:rand).and_return(0)
evaluator.perform
expect(WebMock).to have_requested(:post, endpoint).once
retry_at = 2.hours.from_now
expect(work.reload).to have_attributes(error_code: 'monthly_limit', due_at: retry_at,
full_history_revision: be > work.processed_revision)
expect(ConversationMonitors::Usage.new(account.id).snapshot[:used]).to eq(10_000_000)
create(:message, account: account, conversation: conversation, content: 'Another question during the monthly cooldown')
expect(work.reload).to have_attributes(error_code: 'monthly_limit', due_at: retry_at)
travel_to(reset + 31.minutes)
expect { ConversationMonitors::ProcessJob.perform_now(conversation.id) }.not_to have_enqueued_job(ConversationMonitors::ProcessJob)
expect(WebMock).to have_requested(:post, endpoint).once
expect(ConversationMonitors::Usage.new(account.id).snapshot[:used]).to eq(0)
end
it 'holds an exhausted daily budget until reset without rescheduling on new input' do
reset = Time.current.utc.tomorrow.beginning_of_day
ConversationMonitors::DailyUsage.create!(account: account, usage_date: Time.current.utc.to_date, calls_count: 500_000)
expect { evaluate.call }.to have_enqueued_job(ConversationMonitors::ProcessJob).exactly(:once)
expect(WebMock).not_to have_requested(:post, endpoint)
expect(work.reload).to have_attributes(error_code: 'budget_limit', due_at: be_between(reset, reset + 4.hours),
full_history_revision: be > work.processed_revision)
due_at = work.due_at
6.times { |index| create(:message, account: account, conversation: conversation, content: "During the hold #{index}") }
expect(work.reload).to have_attributes(due_at: due_at, error_code: 'budget_limit', revision: be > work.processed_revision)
expect { ConversationMonitors::ProcessJob.perform_now(conversation.id) }.not_to have_enqueued_job(ConversationMonitors::ProcessJob)
held_request = stub_request(:post, endpoint).with do |request|
JSON.parse(request.body)['state']['messages'].pluck('text').include?('During the hold 0')
end.to_return(status: 200, body: response_body.to_json)
travel_to(work.due_at + 1.second)
evaluate.call
expect(held_request).to have_been_requested.once
expect(work.reload).to have_attributes(due_at: nil, error_code: nil, processed_revision: work.revision)
end
it 'holds input arriving during a budget-rejected request until reset' do
freeze_time
work.update!(due_at: Time.current)
usage = ConversationMonitors::Usage.new(account.id)
allow(ConversationMonitors::Usage).to receive(:new).and_return(usage)
allow(usage).to receive(:reserve!) do
create(:message, account: account, conversation: conversation, content: 'Another question')
raise CustomExceptions::MonitorEvaluationError.new('budget_limit', retry_after: 1.hour.to_i)
end
evaluate.call
expect(work.reload).to have_attributes(error_code: 'budget_limit', due_at: be_between(1.hour.from_now, 5.hours.from_now),
lease_token: nil, revision: be > work.processed_revision)
end
it 'preserves the daily budget hold through a longer earlier batch provider cooldown' do
freeze_time
work.update!(due_at: Time.current)
create_list(:conversation_monitor, 19, account: account, created_at: 1.minute.ago)
stub_request(:post, endpoint).to_return(status: 429, headers: { 'Retry-After' => '7200' })
usage = ConversationMonitors::Usage.new(account.id)
allow(ConversationMonitors::Usage).to receive(:new).and_return(usage)
calls = 0
allow(usage).to receive(:reserve!).and_wrap_original do |original|
calls += 1
raise CustomExceptions::MonitorEvaluationError.new('budget_limit', retry_after: 1.hour.to_i) if calls > 1
original.call
end
evaluator = described_class.new(work.reload)
allow(evaluator).to receive(:rand).and_return(0)
evaluator.perform
expect(WebMock).to have_requested(:post, endpoint).once
retry_at = 2.hours.from_now
expect(work.reload).to have_attributes(error_code: 'budget_limit', due_at: retry_at)
create(:message, account: account, conversation: conversation, content: 'Another question during the provider cooldown')
expect(work.reload).to have_attributes(error_code: 'budget_limit', due_at: retry_at)
travel_to(retry_at - 1.second)
expect { ConversationMonitors::ProcessJob.perform_now(conversation.id) }.not_to have_enqueued_job(ConversationMonitors::ProcessJob)
expect(WebMock).to have_requested(:post, endpoint).once
expect(work.reload).to have_attributes(error_code: 'monthly_limit', due_at: be_within(2.seconds).of(now.beginning_of_month.next_month),
full_history_revision: be > work.processed_revision)
expect(ConversationMonitors::Usage.new(account.id).snapshot[:used]).to eq(100_000)
end
%i[incoming outgoing].each do |message_type|

97
spec/enterprise/services/conversation_monitors/usage_spec.rb
View File

@@ -4,85 +4,76 @@ RSpec.describe ConversationMonitors::Usage do
subject(:usage) { described_class.new(account_id) }
let(:account_id) { create(:account).id }
let(:day) { Time.current.utc.strftime('%Y%m%d') }
let(:token_key) { "conversation_monitors:account:#{account_id}:tokens:#{day}" }
around do |example|
travel_to(Time.utc(2026, 9, 22, 12)) do
with_modified_env(CONVERSATION_MONITORS_DAILY_TOKEN_LIMIT: '1000') { example.run }
end
travel_to(Time.utc(2026, 9, 22, 12)) { example.run }
end
before { Redis::Alfred.delete(token_key) }
after { Redis::Alfred.delete(token_key) }
it 'counts one provider request against the monthly allowance' do
usage.reserve!
it 'reserves one request and reconciles its token usage' do
usage.reserve!(100)
usage.reconcile!(100, 20)
expect(Redis::Alfred.get(token_key).to_i).to eq(20)
expect(Redis::Alfred.ttl(token_key)).to be_positive
expect(ConversationMonitors::DailyUsage.find_by!(account_id: account_id).calls_count).to eq(1)
expect(usage.snapshot).to include(limit: 100_000, used: 1, remaining: 99_999, limit_reached: false, limit_reached_at: nil)
expect(usage.snapshot).to include(limit: 10_000_000, used: 1, remaining: 9_999_999, limit_reached: false, limit_reached_at: nil)
end
it 'allows calls beyond the former account and installation per-minute limits' do
601.times { usage.reserve!(1) }
601.times { usage.reserve! }
expect(Redis::Alfred.get(token_key).to_i).to eq(601)
expect(ConversationMonitors::DailyUsage.find_by!(account_id: account_id).calls_count).to eq(601)
expect(usage.snapshot).to include(used: 601, remaining: 99_399)
expect(usage.snapshot).to include(used: 601, remaining: 9_999_399)
end
it 'releases only the failed reservation and allows a later retry' do
usage.reserve!(100)
it 'does not charge a failed reservation and allows a later retry' do
usage.reserve!
allow(ConversationMonitors::DailyUsage).to receive(:record_call!).and_raise(ActiveRecord::StatementInvalid, 'write failed')
expect { usage.reserve!(300) }.to(raise_error { |error| expect(error.class.name).to eq('ActiveRecord::StatementInvalid') })
expect(Redis::Alfred.get(token_key).to_i).to eq(100)
expect { usage.reserve! }.to(raise_error { |error| expect(error.class.name).to eq('ActiveRecord::StatementInvalid') })
expect(usage.snapshot[:used]).to eq(1)
allow(ConversationMonitors::DailyUsage).to receive(:record_call!).and_call_original
usage.reserve!(300)
usage.reserve!
expect(Redis::Alfred.get(token_key).to_i).to eq(400)
expect(usage.snapshot[:used]).to eq(2)
end
it 'compensates a database rollback after the daily counter was written' do
it 'rolls back the daily counter when the transaction fails' do
allow(ConversationMonitors::DailyUsage).to receive(:record_call!).and_wrap_original do |original, *args, **kwargs|
original.call(*args, **kwargs)
raise ActiveRecord::StatementInvalid, 'transaction failed'
end
expect { usage.reserve!(100) }.to(raise_error { |error| expect(error.class.name).to eq('ActiveRecord::StatementInvalid') })
expect(Redis::Alfred.get(token_key).to_i).to eq(0)
expect { usage.reserve! }.to(raise_error { |error| expect(error.class.name).to eq('ActiveRecord::StatementInvalid') })
expect(ConversationMonitors::DailyUsage.where(account_id: account_id)).not_to exist
end
it 'does not charge the account when the daily token budget is exhausted' do
Redis::Alfred.setex(token_key, 1000, 2.days)
it 'admits the last daily call and blocks further calls until midnight UTC' do
daily = ConversationMonitors::DailyUsage.create!(account_id: account_id, usage_date: Date.current, calls_count: 499_999)
usage.reserve!
expect { usage.reserve!(1) }.to raise_error(CustomExceptions::MonitorEvaluationError) { |error|
expect { usage.reserve! }.to raise_error(CustomExceptions::MonitorEvaluationError) { |error|
expect(error).to have_attributes(code: 'budget_limit', retry_after: 12.hours.to_i)
}
expect(Redis::Alfred.get(token_key).to_i).to eq(1000)
expect(ConversationMonitors::DailyUsage.where(account_id: account_id)).not_to exist
expect(daily.reload).to have_attributes(calls_count: 500_000, limit_reached_at: nil)
expect(usage.snapshot).to include(used: 500_000, remaining: 9_500_000, limit_reached: false)
end
it 'sums daily calls, admits the last credit, and records the timestamp exactly once' do
ConversationMonitors::DailyUsage.create!(account_id: account_id, usage_date: Date.current - 1, calls_count: 99_999)
(2..20).each do |days_ago|
ConversationMonitors::DailyUsage.create!(account_id: account_id, usage_date: Date.current - days_ago, calls_count: 500_000)
end
ConversationMonitors::DailyUsage.create!(account_id: account_id, usage_date: Date.current - 1, calls_count: 499_999,
limit_reached_at: 1.day.ago)
monitor = create(:conversation_monitor, account_id: account_id)
allow(ConversationMonitors::BroadcastJob).to receive(:schedule)
usage.reserve!(10)
usage.reserve!
expect(usage.snapshot).to include(used: 100_000, remaining: 0, limit_reached: true, limit_reached_at: Time.current.to_i)
expect(usage.snapshot[:daily].last(2)).to eq([{ date: '2026-09-21', calls: 99_999 }, { date: '2026-09-22', calls: 1 }])
expect(usage.snapshot).to include(used: 10_000_000, remaining: 0, limit_reached: true, limit_reached_at: Time.current.to_i)
expect(usage.snapshot[:daily].last(2)).to eq([{ date: '2026-09-21', calls: 499_999 }, { date: '2026-09-22', calls: 1 }])
expect(ConversationMonitors::BroadcastJob).to have_received(:schedule).with(monitor.id).once
reached_at = ConversationMonitors::DailyUsage.find_by!(account_id: account_id, usage_date: Date.current).limit_reached_at
expect { usage.reserve!(10) }.to raise_error(CustomExceptions::MonitorEvaluationError) { |error|
expect { usage.reserve! }.to raise_error(CustomExceptions::MonitorEvaluationError) { |error|
expect(error).to have_attributes(code: 'monthly_limit', retry_after: (Time.utc(2026, 10, 1) - Time.current).to_i)
}
expect(ConversationMonitors::DailyUsage.find_by!(account_id: account_id, usage_date: Date.current))
@@ -90,33 +81,41 @@ RSpec.describe ConversationMonitors::Usage do
end
it 'keeps daily history and starts a fresh monthly allowance at midnight UTC' do
old = ConversationMonitors::DailyUsage.create!(account_id: account_id, usage_date: Date.current, calls_count: 100_000,
(1..19).each do |days_ago|
ConversationMonitors::DailyUsage.create!(account_id: account_id, usage_date: Date.current - days_ago, calls_count: 500_000)
end
old = ConversationMonitors::DailyUsage.create!(account_id: account_id, usage_date: Date.current, calls_count: 500_000,
limit_reached_at: Time.current)
travel_to Time.utc(2026, 10, 1)
usage.reserve!(10)
usage.reserve!
expect(usage.snapshot).to include(used: 1, remaining: 99_999, limit_reached: false, limit_reached_at: nil,
expect(usage.snapshot).to include(used: 1, remaining: 9_999_999, limit_reached: false, limit_reached_at: nil,
period_start: '2026-10-01', resets_at: Time.utc(2026, 11, 1).to_i)
expect(old.reload.calls_count).to eq(100_000)
expect(old.reload.calls_count).to eq(500_000)
expect(old.limit_reached_at).to be_present
end
it 'isolates the monthly allowance and daily records between accounts' do
other = create(:account)
ConversationMonitors::DailyUsage.create!(account: other, usage_date: Date.current, calls_count: 100_000, limit_reached_at: Time.current)
(1..19).each do |days_ago|
ConversationMonitors::DailyUsage.create!(account: other, usage_date: Date.current - days_ago, calls_count: 500_000)
end
ConversationMonitors::DailyUsage.create!(account: other, usage_date: Date.current, calls_count: 500_000, limit_reached_at: Time.current)
usage.reserve!(10)
usage.reserve!
expect(usage.snapshot).to include(used: 1, limit_reached: false)
expect(described_class.new(other.id).snapshot).to include(used: 100_000, limit_reached: true)
expect(described_class.new(other.id).snapshot).to include(used: 10_000_000, limit_reached: true)
end
it 'reconciles a request to its reserved UTC day even if it finishes after midnight' do
usage.reserve!(100)
travel 1.day
usage.reconcile!(100, 20)
it 'starts a fresh daily allowance at midnight UTC while keeping the monthly usage' do
daily = ConversationMonitors::DailyUsage.create!(account_id: account_id, usage_date: Date.current, calls_count: 500_000)
expect { usage.reserve! }.to raise_error(CustomExceptions::MonitorEvaluationError, 'budget_limit')
travel_to Time.utc(2026, 9, 23)
usage.reserve!
expect(Redis::Alfred.get(token_key).to_i).to eq(20)
expect(ConversationMonitors::DailyUsage.find_by!(account_id: account_id).usage_date).to eq(Date.new(2026, 9, 22))
expect(daily.reload.calls_count).to eq(500_000)
expect(usage.snapshot).to include(used: 500_001, remaining: 9_499_999, limit_reached: false)
expect(usage.snapshot[:daily].last).to eq(date: '2026-09-23', calls: 1)
end
end