Module: Batchwatch

Defined in:
lib/batchwatch/job.rb,
lib/batchwatch/spool.rb,
lib/batchwatch/client.rb

Overview

The high-level batch job - the client does the annoying parts.

The epic's core promise is "we take the callable, never the payload". Every other batchwatch surface answers a question and hands the decision back to you; this one carries it out. You hand us two callables - the batch-create and the synchronous fallback - and we run them, watch the deadline, poll the batch, and give you a result. We never see or construct the provider payload: there is no field for it, by construction, 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

What happens:

  • job.submit(fn) runs your batch-create callable and remembers the handle it returned and the time it started. We never look inside the handle beyond the questions we have to ask it - is it done, did it succeed, what did it return
    • and those go through callables you can override.
  • job.result waits for the batch. If it finishes before the deadline you get its result. If the deadline arrives first, we cancel the batch (best-effort), run on_deadline and give you that result instead - your job still gets an answer, on time.
  • Every fallback is a measured prediction outcome. A deadline miss is the server's #101 "run_sync" case, so we report it down the SAME accuracy path track() already uses: a completion carrying acted_verdict=false and the deadline_s we were given. No second reporter, no new route, no payload.

The wait itself (#179) owns the poll loop so the caller does not: exponential backoff with jitter, a rate-limit floor so a naive one-second loop against a 24-hour job cannot fire 86,400 requests, a ceiling so the interval does not run away, and a first cadence informed by our own measured p50 for the model. All of it fails open: if batchwatch cannot tell us the p50, we poll on the fixed fallback schedule rather than stopping.

Partial completion (#180) is the third promise: a real batch of 20,000 requests comes back with some landed, some failed per-request, and some never returned before the 24-hour expiry. BatchJob#split hands you a BatchResult that separates those three, mapped back to your own objects by custom_id (never by index - provider ordering is not guaranteed), and BatchJob#retry_failed resubmits only the failed subset with a stable idempotency key so a retry cannot duplicate the job.

Defined Under Namespace

Modules: PollState Classes: AuthError, BatchJob, BatchJobError, BatchResult, Client, HTTPError, Spool, Tracking

Constant Summary collapse

POLL_FLOOR_S =

------------------------------------------------------------- poll cadence

The floor is the rate-limit guard: never poll faster than this, no matter how eager the schedule, because a one-second loop against a 24-hour job is 86,400 requests and an angry provider. The ceiling stops the backoff running away - once the interval reaches it we keep polling at that cadence rather than doubling forever. The base is where the schedule starts when we have no p50 to inform it. All three are seconds; every one is overridable on bw.batch(...).

5.0
POLL_BASE_S =

rate-limit floor: never poll faster than this

5.0
POLL_CEILING_S =

first interval when we have no measured p50

300.0
POLL_BACKOFF =

backoff ceiling: never poll slower than this

2.0
POLL_JITTER =

multiply the interval by this each miss

0.5
POLL_P50_FRACTION =

How much of a model's measured p50 we spend before the first poll. A median queue time of forty minutes means the batch is very unlikely to be done in the first few minutes, so polling then only burns rate limit. We wait a fraction of the p50, clamped to the ceiling, so a slow model gets a patient first poll and a fast one is not starved. This is the use of our own dataset that the card calls a genuine selling point - nobody without the measurements can do it.

0.5
TERMINAL_STATUSES =

Terminal batch statuses across the providers we have seen. A job in one of these is not going to change again, so waiting longer is pointless. Mirrors the status maps in the backfill tool (completed / failed / expired / cancelled), plus OpenAI's "finalizing"-then-"completed" and Anthropic's "ended".

%w[
  completed complete succeeded success
  failed error errored
  expired cancelled canceled ended
].freeze
EXPIRED_STATUSES =

The 24h expiry is a TERMINAL state, not "keep waiting" - and it is a distinct one. Counting an expired job as completed pollutes the accuracy number; counting it as a plain failure loses the signal that the queue was slow. So we name it separately (both here for the poll loop and in #180's split).

%w[expired].freeze
SUCCESS_STATUSES = 
%w[completed complete succeeded success ended].freeze
WHOLE_MARKER =

Markers for the coarse whole-batch split and an unmappable result line. Kept as frozen sentinels rather than magic strings so they cannot collide with a real caller custom_id.

"__batchwatch_whole_batch__"
UNMAPPED_MARKER = 
"__batchwatch_unmapped__"
MAX_BATCH =

The server's cap on one POST /v1/calls/complete.

500
MAX_BYTES =

Cap on the spool file. If batchwatch is down for a week, a busy pipeline must not fill the user's disk. When the cap is reached, the measurement is dropped - and that is the right choice: his machine is not our storage.

5 * 1024 * 1024
VERSION = 
"0.2.2"
SPOOL_INTERVAL_S =

How often, at most, we try to drain the spool on our own.

60.0
ID_WAIT_S =

How long completion waits for the start call's id to land. Only the background thread waits - the caller is long gone.

3.0
ALLOWED_FIELDS =

Field names allowed to leave the machine. Everything else is not in the body. The test test_no_content.rb pins this list. Same fields as the Python, Go, TypeScript and .NET clients.

The last four (acted_verdict, deadline_s, quoted_p50_s, quoted_p90_s) are the outcome measurement (#101): the advice we gave ourselves, attached to the later completion so the SERVER can compare its own measured duration with what we promised. All of it is numbers and a decision we made ourselves - no new PII.

%w[
  provider model mode endpoint requests
  input_tokens output_tokens started_at ended_at
  status ttfb_ms source
  acted_verdict deadline_s quoted_p50_s quoted_p90_s
].freeze

Class Method Summary collapse

Class Method Details

.custom_id_of(item) ⇒ Object

The custom_id of a provider result line, or nil. Duck-typed. Reads item (or [:custom_id], or the attribute) - the field OpenAI and Anthropic both round-trip on a batch line. Nothing else is read.



182
183
184
185
186
187
 
# File 'lib/batchwatch/job.rb', line 182

def self.custom_id_of(item)
  if item.respond_to?(:[])
    return item["custom_id"] || item[:custom_id]
  end
  item.respond_to?(:custom_id) ? item.custom_id : nil
end

.default_cancel(handle) ⇒ Object

Best-effort cancel of a running batch. Never raises upward. Duck-typed against a handle.cancel method if the provider object carries one; otherwise a no-op. The deadline guard has already decided to fall back, so a cancel that fails to land only wastes provider spend on a batch nobody will read.



175
176
177
 
# File 'lib/batchwatch/job.rb', line 175

def self.default_cancel(handle)
  handle.cancel if handle.respond_to?(:cancel)
end

.default_expired(handle) ⇒ Object

Did the batch reach the 24h expiry? Kept separate from default_succeeded so an expiry is a distinct terminal state the caller and the measurement can see - never silently a timeout.



160
161
162
 
# File 'lib/batchwatch/job.rb', line 160

def self.default_expired(handle)
  EXPIRED_STATUSES.include?(handle_status(handle))
end

.default_poll(handle) ⇒ Object

Is this batch finished? Duck-typed against the common provider shape. The OpenAI / Anthropic batch objects both expose a status string; we read it and nothing else. "Finished" means a terminal status; anything else means keep waiting.



148
149
150
 
# File 'lib/batchwatch/job.rb', line 148

def self.default_poll(handle)
  TERMINAL_STATUSES.include?(handle_status(handle))
end

.default_result(handle) ⇒ Object

What the caller gets back for a completed batch: the handle itself. We do not download or parse the batch output - that is the caller's data and their provider SDK's job. Override with result_of: if you want something else.



167
168
169
 
# File 'lib/batchwatch/job.rb', line 167

def self.default_result(handle)
  handle
end

.default_spool ⇒ Object 



76
77
78
79
80
81
82
83
 
# File 'lib/batchwatch/client.rb', line 76

def self.default_spool
  v = ENV["BATCHWATCH_SPOOL"]
  unless v.nil?
    # An empty string turns the spool off.
    return v.empty? ? nil : v
  end
  File.join(Dir.tmpdir, "batchwatch-spool.jsonl")
end

.default_succeeded(handle) ⇒ Object

Did the batch finish successfully (as opposed to failed/expired)?



153
154
155
 
# File 'lib/batchwatch/job.rb', line 153

def self.default_succeeded(handle)
  SUCCESS_STATUSES.include?(handle_status(handle))
end

.default_timeout ⇒ Object 



72
73
74
 
# File 'lib/batchwatch/client.rb', line 72

def self.default_timeout
  (ENV["BATCHWATCH_TIMEOUT"] || "2.0").to_f
end

.default_url ⇒ Object 



68
69
70
 
# File 'lib/batchwatch/client.rb', line 68

def self.default_url
  ENV["BATCHWATCH_URL"] || "https://batchwatch.dev"
end

.handle_status(handle) ⇒ Object

The lower-cased status string of a handle, or "". Duck-typed.

Reads handle.status (a method, or a [:status]/["status"] hash key) and nothing else. Never touches the payload.



134
135
136
137
138
139
140
141
142
 
# File 'lib/batchwatch/job.rb', line 134

def self.handle_status(handle)
  status =
    if handle.respond_to?(:status)
      handle.status
    elsif handle.respond_to?(:[])
      handle[:status] || handle["status"]
    end
  status.nil? ? "" : status.to_s.downcase
end

.idem_complete(group) ⇒ Object

Per-request key for POST /v1/calls/complete. Derived from the group being sent, exactly like backfill.py: a single record is just a group of one. A replay of the SAME records chunks them identically (the spool preserves the order), so the same request carries the same key.



155
156
157
158
159
160
161
162
 
# File 'lib/batchwatch/client.rb', line 155

def self.idem_complete(group)
  return nil if group.nil? || group.empty?

  first = group.first
  last = group.last
  "bw-complete-#{idem_field(first, 'provider')}-#{idem_field(first, 'started_at')}-" \
    "#{group.length}-#{idem_field(last, 'ended_at')}"
end

.idem_start(body) ⇒ Object

Per-record key for POST /v1/calls (the start call).

Keeps the readable provider/model prefix for logs, then a content hash of the full (sanitized) start body. started_at is second-resolution, so two DISTINCT measurements of the same provider+model within one second would otherwise share a key - and the server refuses a reused key that carries a different body (409 body_differs), silently dropping the second measurement. Hashing the whole body keeps distinct measurements apart; an identical body (a retry or a die-and-reflush replay) still hashes the same and dedupes, so #30 holds. This key is only ever computed live at the start POST, never reconstructed from a spool line, so idem_complete's cross-client spool-replay key format is untouched.



146
147
148
149
 
# File 'lib/batchwatch/client.rb', line 146

def self.idem_start(body)
  "bw-start-#{idem_field(body, 'provider')}-#{idem_field(body, 'model')}-" \
    "#{content_hash(body)}"
end

.iso(time) ⇒ Object 



96
97
98
 
# File 'lib/batchwatch/client.rb', line 96

def self.iso(time)
  time.utc.strftime("%Y-%m-%dT%H:%M:%SZ")
end

.outcome_fields(advice) ⇒ Object

The fields from a stored advice that may be attached to a completion (#101).

Only non-nil values come along: if a percentile or a deadline is missing, the field is OMITTED entirely - we do not invent a zero (rule #30). With no advice at all the result is an empty hash, so nothing is attached.



196
197
198
199
200
201
202
203
204
 
# File 'lib/batchwatch/client.rb', line 196

def self.outcome_fields(advice)
  return {} if advice.nil?

  out = { "acted_verdict" => advice["acted_verdict"] }
  %w[deadline_s quoted_p50_s quoted_p90_s].each do |name|
    out[name] = advice[name] unless advice[name].nil?
  end
  out
end

.result_error?(item) ⇒ Boolean

Did this per-request line fail? Duck-typed against the provider shape. A batch output line carries an error (non-null) on failure, or a response with a non-2xx status_code. We read only those envelope fields - never the body.

Returns:

  • (Boolean) 


192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
 
# File 'lib/batchwatch/job.rb', line 192

def self.result_error?(item)
  error =
    if item.respond_to?(:[])
      item["error"] || item[:error]
    elsif item.respond_to?(:error)
      item.error
    end
  return true if error

  response =
    if item.respond_to?(:[])
      item["response"] || item[:response]
    elsif item.respond_to?(:response)
      item.response
    end
  if response.respond_to?(:[])
    code = response["status_code"] || response[:status_code]
    unless code.nil?
      begin
        n = Integer(code)
        return !(n >= 200 && n < 300)
      rescue ArgumentError, TypeError
        return false
      end
    end
  end
  false
end

.sanitize(body) ⇒ Object

Drop everything not on the allowlist.

The last stop before the network. Even though no public method accepts free text, THIS function is the place you can point to when someone asks "how do you know a prompt cannot get out".



90
91
92
93
94
 
# File 'lib/batchwatch/client.rb', line 90

def self.sanitize(body)
  body.each_with_object({}) do |(k, v), out|
    out[k.to_s] = v if ALLOWED_FIELDS.include?(k.to_s)
  end
end

.seconds(max_wait) ⇒ Object

Translate a max_wait into seconds, or nil if we cannot make sense of it.

The caller typically writes "15m", "1h", "30s" or "2d" - the same language /v1/should-i-batch itself accepts. A bare number is read as seconds. If we cannot parse it, we send nil rather than a guess (rule #30).



169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
 
# File 'lib/batchwatch/client.rb', line 169

def self.seconds(max_wait)
  return nil if max_wait.nil?
  return max_wait.to_f if max_wait.is_a?(Numeric)

  s = max_wait.to_s.strip.downcase
  return nil if s.empty?

  factor = { "s" => 1.0, "m" => 60.0, "h" => 3600.0, "d" => 86_400.0 }
  unit = s[-1]
  if factor.key?(unit)
    num = s[0..-2]
    mult = factor[unit]
  else
    num = s
    mult = 1.0
  end
  num = num.strip
  return nil unless num.match?(/\A-?\d+(\.\d+)?\z/)

  num.to_f * mult
end

.track(model, **kw, &blk) ⇒ Object

Shortcut that uses a shared default client, to get started quickly.



826
827
828
829
 
# File 'lib/batchwatch/client.rb', line 826

def self.track(model, **kw, &blk)
  @default_lock.synchronize { @default ||= Client.new }
  @default.track(model, **kw, &blk)
end