Class: Batchwatch::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/batchwatch/client.rb

Overview

The client. Every submission is non-blocking and fails open.

Args:

token: API key. Falls back to $BATCHWATCH_TOKEN. Optional for
measurement, required to replay a spool.
base_url: Default $BATCHWATCH_URL or https://batchwatch.dev
timeout: Seconds per HTTP call. Default $BATCHWATCH_TIMEOUT or 2.0.
enabled: false turns every network call into a no-op.
spool: Path to the spool file, nil to turn it off. Default
$BATCHWATCH_SPOOL or a file in the temp directory. Spooling is inactive
without a token, because the replay route requires one.
logger: Optional logger. All swallowed errors go to logger.debug.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(token: nil, base_url: nil, timeout: nil, enabled: true, spool: UNSET, logger: nil) ⇒ Client

Returns a new instance of Client.



246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
 
# File 'lib/batchwatch/client.rb', line 246

def initialize(token: nil, base_url: nil, timeout: nil, enabled: true,
               spool: UNSET, logger: nil)
  @token = token || ENV["BATCHWATCH_TOKEN"]
  @base = (base_url || Batchwatch.default_url).sub(%r{/+\z}, "")
  @timeout = timeout.nil? ? Batchwatch.default_timeout : timeout
  @enabled = enabled
  @logger = logger
  @threads = []
  @threads_lock = Mutex.new

  path = spool.equal?(UNSET) ? Batchwatch.default_spool : spool
  # Without a key /v1/calls/complete cannot accept it, so a spool file could
  # never be sent. So we do not write it.
  @spool = (path && @token) ? Spool.new(path, logger: logger) : nil
  if path && !@token
    debug("batchwatch: no token - spool disabled " \
          "(/v1/calls/complete requires a key)")
  end
  @spool_last = 0.0
  @spool_lock = Mutex.new
  # Last advice per model (#101). When should_batch answers, we store here
  # what we recommended + the quoted percentiles, so the NEXT completion for
  # the same model can attach them. The correlation is a documented
  # approximation: latest-advice-per-model, not per-job.
  @advice = {}
  @advice_lock = Mutex.new
end

Instance Attribute Details

#base ⇒ Object (readonly)

Returns the value of attribute base.



241
242
243
 
# File 'lib/batchwatch/client.rb', line 241

def base
  @base
end

#enabled ⇒ Object (readonly)

Returns the value of attribute enabled.



241
242
243
 
# File 'lib/batchwatch/client.rb', line 241

def enabled
  @enabled
end

#spool ⇒ Object (readonly)

Returns the value of attribute spool.



241
242
243
 
# File 'lib/batchwatch/client.rb', line 241

def spool
  @spool
end

#timeout ⇒ Object (readonly)

Returns the value of attribute timeout.



241
242
243
 
# File 'lib/batchwatch/client.rb', line 241

def timeout
  @timeout
end

#token ⇒ Object (readonly)

Returns the value of attribute token.



241
242
243
 
# File 'lib/batchwatch/client.rb', line 241

def token
  @token
end

Instance Method Details

#advice(model, max_wait: nil, provider: "openai", input_tokens: nil, output_tokens: nil, max_tokens: nil, risk: "p90") ⇒ Object

The whole verdict. Returns nil if we cannot answer.

output_tokens is usually UNKNOWN at this point - the model decides them. The default is therefore nil, not zero. Sending zero would make the server compute the saving on zero output, and output costs five to six times as much as input: the answer would be systematically too low, without anyone seeing it. If you know a ceiling, pass max_tokens.



374
375
376
377
378
379
380
381
382
383
384
385
386
387
 
# File 'lib/batchwatch/client.rb', line 374

def advice(model, max_wait: nil, provider: "openai",
           input_tokens: nil, output_tokens: nil, max_tokens: nil,
           risk: "p90")
  q = { "provider" => provider, "model" => model, "risk" => risk }
  { "input_tokens" => input_tokens, "output_tokens" => output_tokens,
    "max_tokens" => max_tokens }.each do |name, value|
    q[name] = value.to_i unless value.nil?
  end
  q["max_wait"] = max_wait unless max_wait.nil?
  call("/v1/should-i-batch?#{URI.encode_www_form(q)}")
rescue StandardError => e
  debug("batchwatch advice failed: #{e}")
  nil
end

#batch(model, deadline: nil, on_deadline: nil, provider: "openai", **kw) ⇒ Object

A high-level batch job: deadline guard (#178) + poll loop (#179) + partial completion (#180).

The epic's promise: we take the callable, never the payload. You give us two callables - the batch-create and the synchronous fallback - and we run them, watch the deadline, poll the batch, and hand back a result. We never see or construct the provider payload, exactly as with track().

job = bw.batch("gpt-5.6-sol", deadline: "15m",
               on_deadline: -> { client.chat.completions.create(...) })
job.submit(-> { client.batches.create(...) })
result = job.result

If the batch finishes before deadline you get its result. If the deadline arrives first we cancel the batch (best-effort), run on_deadline and return that instead - so the caller's job always gets an answer, on time. Every fallback is reported down the existing #101 accuracy path (acted_verdict=false on a completion) so the server can score the prediction; nothing new is sent.

See Batchwatch::BatchJob for the full contract (poll cadence, poll_once, split, retry_failed, and the injectable clock/sleep/rng seams).



492
493
494
495
 
# File 'lib/batchwatch/client.rb', line 492

def batch(model, deadline: nil, on_deadline: nil, provider: "openai", **kw)
  BatchJob.new(self, model, deadline: deadline, on_deadline: on_deadline,
                          provider: provider, **kw)
end

#call(path, method: "GET", body: nil, timeout: nil, idem: nil) ⇒ Object

One HTTP call. Raises on error - only public methods swallow.

idem is an optional Idempotency-Key. On the write routes (/v1/calls and /v1/calls/complete) it lets the server dedupe a resend: the same body + the same key writes the measurement once, no matter how many times a die-and-reflush sends it (#30). The key is derived from the measurement, so the replay carries the same key as the first attempt. nil = no header.

Raises:



283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
 
# File 'lib/batchwatch/client.rb', line 283

def call(path, method: "GET", body: nil, timeout: nil, idem: nil)
  uri = URI.parse(@base + path)
  deadline = timeout || @timeout

  req_class = case method
              when "GET" then Net::HTTP::Get
              when "POST" then Net::HTTP::Post
              when "PATCH" then Net::HTTP::Patch
              when "DELETE" then Net::HTTP::Delete
              else raise ArgumentError, "unknown method #{method}"
              end
  req = req_class.new(uri)
  req["user-agent"] = "batchwatch-ruby/#{VERSION}"
  unless body.nil?
    req["content-type"] = "application/json"
    req.body = JSON.generate(body)
  end
  req["authorization"] = "Bearer #{@token}" if @token
  req["idempotency-key"] = idem if idem && !idem.empty?

  http = Net::HTTP.new(uri.host, uri.port)
  http.use_ssl = (uri.scheme == "https")
  # Both the connect and the read timeout, so a server that accepts but
  # never answers cannot hold us longer than the deadline.
  http.open_timeout = deadline
  http.read_timeout = deadline

  res = http.request(req)
  code = res.code.to_i
  raw = res.body || ""
  raise HTTPError.new(code, raw) if code < 200 || code >= 300

  raw.strip.empty? ? nil : JSON.parse(raw)
end

#flush(timeout: 5.0) ⇒ Object

Wait for outstanding submissions. Call it before the process exits.



342
343
344
345
346
347
348
349
350
351
 
# File 'lib/batchwatch/client.rb', line 342

def flush(timeout: 5.0)
  deadline = monotonic + timeout
  snapshot = @threads_lock.synchronize { @threads.dup }
  snapshot.each do |t|
    rest = deadline - monotonic
    t.join(rest > 0 ? rest : 0)
  end
  @threads_lock.synchronize { @threads.reject! { |t| !t.alive? } }
  @threads_lock.synchronize { @threads.empty? }
end

#flush_spool(timeout: nil) ⇒ Object

Send everything waiting on disk. Returns the number accepted.

Synchronous and safe to call from a shutdown hook. Never raises. Records the server rejects as invalid are dropped - they will never become valid

  • and the count is logged.


606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
 
# File 'lib/batchwatch/client.rb', line 606

def flush_spool(timeout: nil)
  return 0 if !@spool || !@enabled

  records = @spool.take
  return 0 if records.empty?

  sent = 0
  rest = records.dup
  until rest.empty?
    group = rest.shift(MAX_BATCH)
    begin
      # The key is derived from the group itself: a replay after a crash
      # chunks the records identically (the spool preserves the order), so
      # the key is the same and the server dedupes the double write (#30).
      r = call("/v1/calls/complete", method: "POST", body: group,
                                     timeout: timeout || [@timeout, 10.0].max,
                                     idem: Batchwatch.idem_complete(group))
    rescue StandardError => e
      debug("batchwatch: spool could not be sent: #{e}")
      # The group that failed stays put along with the rest.
      @spool.keep(group + rest)
      return sent
    end
    sent += (r || {}).fetch("accepted", 0)
    rejected = (r || {}).fetch("rejected", 0)
    if rejected && rejected > 0
      debug("batchwatch: #{rejected} spooled measurements were rejected and dropped")
    end
  end
  @spool.keep([])
  debug("batchwatch: #{sent} spooled measurements sent")
  sent
end

#get_advice(model) ⇒ Object

The latest advice for a model, or nil. The approximation is latest-advice-per-model: the newest should_batch attaches to the next completion of the same model.



427
428
429
 
# File 'lib/batchwatch/client.rb', line 427

def get_advice(model)
  @advice_lock.synchronize { @advice[model] }
end

#in_background(&block) ⇒ Object

Run on a background thread. Errors are swallowed - this must never take the caller's own call down with it.



320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
 
# File 'lib/batchwatch/client.rb', line 320

def in_background(&block)
  return nil unless @enabled

  t = Thread.new do
    begin
      block.call
    rescue HTTPError => e
      debug("batchwatch #{e.status}: #{e.body}")
    rescue StandardError => e
      debug("batchwatch unavailable: #{e}")
    end
  end
  # Clean up finished threads, so the list does not grow without bound in a
  # long-running process.
  @threads_lock.synchronize do
    @threads.reject! { |x| !x.alive? }
    @threads << t
  end
  t
end

#key_status ⇒ Object

THIS key's tier, contribution status and quota. Requires a key.

GET /v1/keys/current returns the row verbatim: "tier", "contributing", "recent_measurements", "required", "window_days", "delayed_by_s", "live", "quota": {...}. tier is the effective tier derived from your measurements (free / contributor; only paid is operator-assigned), and quota reports the calls window for the gated routes.

Unlike the measurement path this does NOT fail open: without a key there is no key to describe, so we raise AuthError.



594
595
596
597
 
# File 'lib/batchwatch/client.rb', line 594

def key_status
  require_key("key_status")
  call("/v1/keys/current")
end

#maybe_flush_spool ⇒ Object

Called from the background thread AFTER a successful call - so we know the network is up right now, and we avoid hammering a server that is not answering anyway.



643
644
645
646
647
648
649
650
651
652
653
654
655
 
# File 'lib/batchwatch/client.rb', line 643

def maybe_flush_spool
  return unless @spool

  now = monotonic
  @spool_lock.synchronize do
    return if now - @spool_last < SPOOL_INTERVAL_S

    @spool_last = now
  end
  flush_spool
rescue StandardError => e
  debug("batchwatch: spool drain failed: #{e}")
end

#my_calls(after: nil, limit: nil) ⇒ Object

The measurements THIS key has contributed. Requires a key.

GET /v1/calls/mine returns everything the service holds that came from your key - the only readback there is: there is no route to a single call by id, and none to anyone else's rows. Use it to confirm a measurement landed after track() / flush_spool().

Returns the server's row verbatim: "count", "next", "calls": [...], "note". next is a ready-made URL for the following page (null on the last), keyed on started_server so a walk cannot skip a row that arrives mid-walk.

after: unix seconds; only rows with a later started_server come back.
Omit for the first page.
limit: rows per page, clamped server-side to 1-1000 (default 500).

Unlike the measurement path this does NOT fail open: the route is per-key, so without one we raise AuthError rather than pretend.



573
574
575
576
577
578
579
580
581
 
# File 'lib/batchwatch/client.rb', line 573

def my_calls(after: nil, limit: nil)
  require_key("my_calls")
  q = {}
  q["after"] = after.to_i unless after.nil?
  q["limit"] = limit.to_i unless limit.nil?
  path = "/v1/calls/mine"
  path += "?#{URI.encode_www_form(q)}" unless q.empty?
  call(path)
end

#record_job_outcome(model, provider, acted_verdict, deadline_s) ⇒ Object

Report a deadline-guard outcome down the existing #101 path.

The deadline guard is a should_batch-style prediction carried out: when it falls back, the outcome is the server's "run_sync" case ("batching would have missed"). #101 already carries exactly that - a completion measurement bearing acted_verdict and deadline_s, which the server judges against its own measured duration and rolls into /v1/accuracy. So we seed the same per-model advice slot should_batch writes and emit a completion through the SAME track() path - no second reporter, no new route.

A completion in time carries acted_verdict=true (the batch held the deadline); a fallback carries false. Both sides are reported, which is what keeps the accuracy number honest. Percentiles are omitted, not invented as zero (rule #30): the guard did not quote a p90.



511
512
513
514
515
516
517
518
519
520
521
 
# File 'lib/batchwatch/client.rb', line 511

def record_job_outcome(model, provider, acted_verdict, deadline_s)
  @advice_lock.synchronize do
    @advice[model] = {
      "acted_verdict" => acted_verdict,
      "deadline_s" => deadline_s,
      "quoted_p50_s" => nil,
      "quoted_p90_s" => nil
    }
  end
  track(model, provider: provider, mode: "batch") { |t| t.done }
end

#remember_advice(model, acted_verdict, max_wait, answer) ⇒ Object

Store the advice we just gave for THIS model (#101). Only numbers and our own decision end up here - no user data. The quoted percentiles are taken from the server's response; if one of them is missing, we store nil and simply do not attach it (rule #30: nothing invented as 0).



413
414
415
416
417
418
419
420
421
422
 
# File 'lib/batchwatch/client.rb', line 413

def remember_advice(model, acted_verdict, max_wait, answer)
  num = ->(v) { v.is_a?(Numeric) ? v.to_f : nil }
  advice = {
    "acted_verdict" => acted_verdict,
    "deadline_s" => Batchwatch.seconds(max_wait),
    "quoted_p50_s" => num.call(answer["p50_s"]),
    "quoted_p90_s" => num.call(answer["p90_s"])
  }
  @advice_lock.synchronize { @advice[model] = advice }
end

#should_batch(model, max_wait: nil, default: false, **kw) ⇒ Object

true/false. On any doubt you get your default back.

We never guess on the caller's behalf: if we cannot answer, the caller gets their own predetermined value. The default is false - "run it synchronously" - which is the safe way to be wrong, because a synchronous call just costs more, while an unexpected eight-hour queue can take a product down.



396
397
398
399
400
401
402
403
404
405
406
407
 
# File 'lib/batchwatch/client.rb', line 396

def should_batch(model, max_wait: nil, default: false, **kw)
  r = advice(model, max_wait: max_wait, **kw)
  return default unless r

  answer = case r["verdict"]
           when "run_batch" then true
           when "run_sync", "batch_at" then false
           else default
           end
  remember_advice(model, answer, max_wait, r)
  answer
end

#spool_measurement(body) ⇒ Object

Store a COMPLETED measurement that could not be delivered.



658
659
660
661
662
663
664
 
# File 'lib/batchwatch/client.rb', line 658

def spool_measurement(body)
  if @spool
    @spool.append(body)
  else
    debug("batchwatch: the measurement was lost (no spool)")
  end
end

#subscribe(channel, target, secret: nil, providers: nil, models: nil, min_severity: nil) ⇒ Object

Subscribe to outcome alerts for THIS key. Returns the row.

channel is "webhook" or "slack", target an https URL. secret applies only to webhook: if you omit it, the server makes one itself and returns the plaintext ONCE in secret - read it here and configure your receiver, it never comes again. providers and models are comma-strings (or omit

all), min_severity is "degraded" or "severe".

Unlike the measurement path, this one does NOT fail open: subscribing is an explicit action against an authorized route, so without a key we raise AuthError rather than silently pretend.



444
445
446
447
448
449
450
451
452
453
 
# File 'lib/batchwatch/client.rb', line 444

def subscribe(channel, target, secret: nil, providers: nil, models: nil,
              min_severity: nil)
  require_key("subscribe")
  body = { "channel" => channel, "target" => target }
  { "secret" => secret, "providers" => providers, "models" => models,
    "min_severity" => min_severity }.each do |name, value|
    body[name] = value unless value.nil?
  end
  call("/v1/subscriptions", method: "POST", body: body)
end

#subscriptions ⇒ Object

THIS key's active subscriptions, as an array. Never the secret.



456
457
458
459
460
 
# File 'lib/batchwatch/client.rb', line 456

def subscriptions
  require_key("subscriptions")
  r = call("/v1/subscriptions")
  (r || {}).fetch("subscriptions", [])
end

#track(model, provider: "openai", mode: "batch", requests: 1, input_tokens: nil, endpoint: nil) ⇒ Object

Measure one call. The submission happens in the background.

No exception from batchwatch ever reaches the caller. An exception from the caller's own block is recorded as "failed" and re-raised untouched.

With a block it acts like the Python context manager: it closes the measurement for you, including on the error path. Without a block it returns the tracking handle and you call done() yourself.



533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
 
# File 'lib/batchwatch/client.rb', line 533

def track(model, provider: "openai", mode: "batch", requests: 1,
          input_tokens: nil, endpoint: nil)
  t = Tracking.new(self, provider, model, mode, requests, input_tokens,
                   endpoint)
  t.start
  return t unless block_given?

  begin
    yield t
  rescue Exception # rubocop:disable Lint/RescueException
    # Everything - including Interrupt/SignalException - is recorded as
    # failed and re-raised untouched. We swallow our own errors, never the
    # caller's.
    t.done(status: "failed")
    raise
  else
    t.done unless t.finished?
  end
end

#unsubscribe(sub_id) ⇒ Object

Revoke one of your OWN subscriptions by numeric id. Returns the response.



463
464
465
466
 
# File 'lib/batchwatch/client.rb', line 463

def unsubscribe(sub_id)
  require_key("unsubscribe")
  call("/v1/subscriptions/#{sub_id}", method: "DELETE")
end

#wait_now(model, provider: "openai", mode: "batch") ⇒ Object

What the queue is doing right now, or nil if we cannot say.



356
357
358
359
360
361
362
363
364
365
 
# File 'lib/batchwatch/client.rb', line 356

def wait_now(model, provider: "openai", mode: "batch")
  q = URI.encode_www_form(provider: provider, model: model, mode: mode)
  r = call("/v1/wait?#{q}")
  return nil if !r || r["verdict"] == "insufficient_data"

  r
rescue StandardError => e
  debug("batchwatch wait failed: #{e}")
  nil
end