# Batch Mode

Providers sell the same model twice. A live call answers in seconds at list
price; the same request submitted to a **Message Batches** API answers within
24 hours at roughly half the token cost. For a nightly enrichment run, an
overnight backfill, or any step whose result nobody is waiting on, that is a
50% discount for latency you weren't spending anyway.

Baton makes that a one-line change to a step:

```elixir
use Baton.LLMStep, mode: :batch
```

Everything else about the step is unchanged — same `request/1`, same
`handle_response/3`, same decoding and error taxonomy. Downstream steps,
completion, retries, and cost stats behave exactly as they do for a live step.
See the [LLM step basics](../README.md#llm-steps) first if you haven't written
one yet.

## A complete example

Here is an ordinary live step:

```elixir
defmodule MyApp.Steps.Classify do
  use Baton.LLMStep

  @impl true
  def output_schema do
    %{
      "type" => "object",
      "properties" => %{"category" => %{"type" => "string"}},
      "required" => ["category"]
    }
  end

  @impl true
  def request(%Oban.Job{} = job) do
    {:ok, %{"text" => text}} = Baton.Results.get_result(job, :extract)

    messages = [%{role: "user", content: "Classify this document:\n\n#{text}"}]
    {:ok, messages, model: "claude-sonnet-4-20250514", max_tokens: 1024}
  end

  @impl true
  def handle_response(%{"category" => category}, _context, _job) do
    {:ok, %{"category" => category}}
  end
end
```

And here it is on the batch transport. The diff is the `use` line:

```elixir
defmodule MyApp.Steps.Classify do
  use Baton.LLMStep, mode: :batch

  # ... output_schema/0, request/1, and handle_response/3 unchanged ...
end
```

It goes into a workflow like any other step, and its dependents don't know the
difference:

```elixir
Baton.new(workflow_name: "nightly-enrichment")
|> Baton.add(:extract,  MyApp.Steps.Extract.new(%{doc_id: id}))
|> Baton.add(:classify, MyApp.Steps.Classify.new(%{}), deps: [:extract])
|> Baton.add(:store,    MyApp.Steps.Store.new(%{}),    deps: [:classify])
|> Baton.insert!()
```

`:store` waits for `:classify` exactly as it would for a live step — it just
waits hours instead of seconds, then reads the same result map.

## What the engine does

A batch step runs as several short Oban attempts rather than one long one:

1. **Submit.** `request/1` builds the messages, the engine gates on the rate
   limiter, submits a one-request batch, records the batch id on the step's
   node, and snoozes.
2. **Poll.** Each later attempt reads the batch id and asks the provider
   whether the batch has ended. If not, it snoozes again — no dependency reads,
   no prompt rendering, nothing sent.
3. **Ingest.** Once the batch ends, the engine collects the result and runs it
   through the same pipeline a live response takes: `decode/1` →
   `handle_response/3` → attach usage.

> #### Waiting is free {: .info}
>
> Oban raises `max_attempts` alongside `attempt` on every snooze, so polling
> never eats the retry budget: a step can poll for a day and still have its
> full quota of attempts left for real failures. A snoozing step also sits in
> `scheduled`, which keeps the workflow open and its dependents parked.

## The client contract

Batch mode needs three more functions from your `:llm_client` than the live
path does. They're optional callbacks on `Baton.LLMClient`:

```elixir
defmodule MyApp.LLM.Client do
  @behaviour Baton.LLMClient

  @impl true
  def complete(messages, opts) do
    # ... the live path you already have ...
  end

  @impl true
  def submit_batch(requests, opts) do
    # requests: [%{custom_id: "job-4171", messages: [...], opts: [...]}]
    # Return the provider's batch id.
    {:ok, "msgbatch_01ABC..."}
  end

  @impl true
  def poll_batch(batch_id, _opts) do
    # {:ok, :pending} keeps the step snoozing; {:ok, :ended} moves it on.
    # "Ended" means the provider is done, not that the request succeeded.
    {:ok, :ended}
  end

  @impl true
  def batch_results(batch_id, _opts) do
    {:ok,
     [
       %{
         custom_id: "job-4171",
         type: :succeeded,       # or :errored | :canceled | :expired
         response: %{text: "...", model: "...", stop_reason: "end_turn", usage: %{...}},
         error: nil
       }
     ]}
  end
end
```

`submit_batch/2` takes a **list** even though Baton submits one request per
step today — provider APIs are list-shaped, and a future coalescer that packs
many steps into one batch will reuse the callback unchanged.

Baton matches its result out of the list by `custom_id`, which it sets to
`"job-<oban job id>"`: stable across attempts and short enough for any
provider's limit.

If your client doesn't implement these, a batch step **cancels** on its first
attempt with `{:batch_unsupported, MyApp.LLM.Client}` rather than retrying
against a gap no retry can close.

That check is module-wide, so it can't see a failure that depends on the
*call*: a model routed to a provider whose batch API you haven't implemented,
a missing credential, an endpoint you can't speak. Those have no HTTP status
to report and no retry that fixes them, so say so directly rather than
inventing a status you never received:

```elixir
{:error, {:cancel, {:batch_unsupported_provider, "openai"}}}
```

The taxonomy honours the verdict and cancels on the first attempt. Anything
else keeps its normal retry budget.

> #### Return atom keys {: .warning}
>
> Baton looks results up by `:custom_id` and reads `response.text`,
> `response.stop_reason`, `response.model`, and `response.usage` — all atom
> keys. A client that hands back string-keyed maps decoded straight from JSON
> will look like it found nothing and fail the step with
> `:batch_result_missing`.

## Example: an OpenAI batch client

Nothing in Baton is provider-specific — it never speaks HTTP, and the three
callbacks are all it knows. Anthropic's batch API takes requests inline;
OpenAI's is file-based (upload a JSONL, create a batch from the file, download
an output file), which is more work but fits the same contract. Here is the
whole of it, using [Req](https://hex.pm/packages/req) for HTTP:

```elixir
defmodule MyApp.LLM.OpenAIBatch do
  @behaviour Baton.LLMClient

  @base "https://api.openai.com/v1"
  @endpoint "/v1/chat/completions"

  # Statuses that mean "still working". `cancelling` belongs here: it has not
  # settled yet, and polling again is how we find out that it did.
  @pending ~w(validating in_progress finalizing cancelling)

  @impl true
  def complete(messages, opts), do: MyApp.LLM.OpenAI.complete(messages, opts)

  @impl true
  def submit_batch(requests, _opts) do
    jsonl =
      Enum.map_join(requests, "\n", fn request ->
        Jason.encode!(%{
          custom_id: request.custom_id,
          method: "POST",
          url: @endpoint,
          body: %{
            model: request.opts[:model],
            max_tokens: request.opts[:max_tokens],
            messages: request.messages
          }
        })
      end)

    with {:ok, %{"id" => file_id}} <- upload(jsonl),
         {:ok, %{"id" => batch_id}} <- create_batch(file_id) do
      {:ok, batch_id}
    end
  end

  @impl true
  def poll_batch(batch_id, _opts) do
    case get_batch(batch_id) do
      {:ok, %{"status" => status}} when status in @pending -> {:ok, :pending}
      {:ok, %{"status" => _settled}} -> {:ok, :ended}
      {:error, _} = error -> error
    end
  end

  @impl true
  def batch_results(batch_id, _opts) do
    with {:ok, batch} <- get_batch(batch_id) do
      # Successes and failures come back in two separate files; Baton wants one
      # list, so read both. Either may be absent.
      results =
        [batch["output_file_id"], batch["error_file_id"]]
        |> Enum.reject(&is_nil/1)
        |> Enum.flat_map(&download_lines/1)
        |> Enum.map(&to_result/1)

      {:ok, results}
    end
  end

  # ── HTTP ──────────────────────────────────────────────────────────────────

  defp upload(jsonl) do
    Req.post("#{@base}/files",
      auth: {:bearer, api_key()},
      form_multipart: [purpose: "batch", file: {jsonl, filename: "batch.jsonl"}]
    )
    |> unwrap()
  end

  defp create_batch(file_id) do
    Req.post("#{@base}/batches",
      auth: {:bearer, api_key()},
      json: %{input_file_id: file_id, endpoint: @endpoint, completion_window: "24h"}
    )
    |> unwrap()
  end

  defp get_batch(batch_id) do
    Req.get("#{@base}/batches/#{batch_id}", auth: {:bearer, api_key()}) |> unwrap()
  end

  defp download_lines(file_id) do
    case Req.get("#{@base}/files/#{file_id}/content", auth: {:bearer, api_key()}) do
      {:ok, %{status: 200, body: body}} ->
        body |> String.split("\n", trim: true) |> Enum.map(&Jason.decode!/1)

      _ ->
        []
    end
  end

  # Failures are reported as `%{status: n}` so Baton's error taxonomy can read
  # them: a 429 while submitting becomes a snooze rather than a burned attempt.
  defp unwrap({:ok, %{status: 200, body: body}}), do: {:ok, body}
  defp unwrap({:ok, %{status: status, body: body}}), do: {:error, %{status: status, body: body}}
  defp unwrap({:error, reason}), do: {:error, reason}

  defp api_key, do: Application.fetch_env!(:my_app, :openai_api_key)

  # ── Shape mapping ─────────────────────────────────────────────────────────

  defp to_result(%{"custom_id" => id, "response" => %{"status_code" => 200, "body" => body}}) do
    %{custom_id: id, type: :succeeded, response: normalize(body), error: nil}
  end

  # Hand the failure back as `%{status: n}` — the shape `classify_error/1`
  # reads. That is what makes a 400 cancel the step (resubmitting the same
  # request would fail identically) while a 429 only snoozes it.
  defp to_result(%{"custom_id" => id, "response" => %{"status_code" => status} = response}) do
    %{
      custom_id: id,
      type: :errored,
      response: nil,
      error: %{status: status, body: response["body"]}
    }
  end

  defp to_result(%{"custom_id" => id} = line) do
    %{custom_id: id, type: :errored, response: nil, error: line["error"]}
  end

  defp normalize(%{"choices" => [choice | _], "model" => model, "usage" => usage}) do
    cached = get_in(usage, ["prompt_tokens_details", "cached_tokens"]) || 0

    %{
      text: get_in(choice, ["message", "content"]),
      model: model,
      stop_reason: stop_reason(choice["finish_reason"]),
      usage: %{
        # OpenAI's prompt_tokens *includes* cached tokens; the convention Baton
        # records and reconciles against is Anthropic's, where input_tokens
        # excludes them. Subtract, or every cache read is counted twice.
        input_tokens: usage["prompt_tokens"] - cached,
        output_tokens: usage["completion_tokens"],
        cache_read_input_tokens: cached
      }
    }
  end

  # "length" is OpenAI's truncation signal. It has to become "max_tokens" or
  # the check that makes a truncated reply retryable never fires.
  defp stop_reason("length"), do: "max_tokens"
  defp stop_reason(other), do: other
end
```

Three things in there are easy to get wrong and worth calling out:

- **`finish_reason: "length"` must become `stop_reason: "max_tokens"`.** Baton
  matches that exact string to turn a truncated reply into a retryable error.
  Skip it and a half-finished response gets decoded as if it were complete.
- **`prompt_tokens` double-counts cache reads** relative to the convention
  Baton's stats and rate reconciliation use. Subtract `cached_tokens` out of it.
- **A `failed` batch (input validation) produces no output file.** Mapping it
  to `{:ok, :ended}`, as above, means Baton finds no result for its `custom_id`
  and fails the step with `:batch_result_missing` — correct, if terse. Return
  `{:error, ...}` from `poll_batch/2` instead if you would rather the step
  retry, but note a retry resubmits the same request, which will fail the same
  way.

The first two apply equally to a live OpenAI client, so if you already have a
working `complete/2` you likely have them solved and can share the mapping.

## Tuning

| Option | Default | Meaning |
|---|---|---|
| `:poll_interval` | `300` (5 min) | Seconds to snooze between polls |
| `:batch_deadline` | `90_000` (25 h) | Give up if the batch hasn't ended |

```elixir
use Baton.LLMStep, mode: :batch, poll_interval: 600
```

The poll interval trades wasted wake-ups against how long a finished batch sits
unnoticed; five minutes is negligible against a multi-hour turnaround. The
deadline is a backstop *above* the provider's own 24-hour expiry — reaching it
means something is wrong with the handle, not the batch.

## In a portable flow

`use Baton.LLMStep, mode: :batch` fixes the transport when the module compiles,
which is right for a step module that exists to do one thing. A portable flow
node can't work that way: every `llm` node in every definition runs through the
single `Baton.Flow.Workers.LLM`. So a flow node names its transport in `config`
instead, and the worker reads it on each attempt:

```elixir
%Baton.Flow.NodeSpec{
  id: "classify",
  type: "llm",
  config: %{
    "model" => "claude-sonnet-4-20250514",
    "user_prompt" => %{"body" => "..."},
    "transport" => "batch",
    "poll_interval" => 600
  }
}
```

`"transport"` is `"live"` (the default) or `"batch"`; `"poll_interval"` and
`"batch_deadline"` are the same seconds-valued options as above, and anything
you leave out falls back to the engine's defaults. Nothing else about the node
changes — prompts, bindings, schemas, adapters, and the results downstream
nodes read are all identical.

`Baton.Flow.Validator` checks this before a definition compiles, because every
way of getting it wrong fails *silently* otherwise — an unrecognized transport
would just run live, and you'd find out from the bill. It rejects:

| Error | Cause |
|---|---|
| `:unsupported` | a transport that isn't `"live"` or `"batch"` |
| `:tuning_without_batch` | `poll_interval`/`batch_deadline` on a node that isn't batched — usually a typo in `transport` |
| `:not_an_llm_node` | transport keys on an `action` node, which has no transport |
| `:invalid_poll_interval` / `:invalid_batch_deadline` | not a positive integer of seconds |
| `:sequential_fan_out` | `"batch"` on a fan-out gated `sequential` |

That last one is a judgment call worth explaining. The `sequential` gate chains
expanded nodes so each waits for the one before it. Batched, that is N waits of
up to 24 hours apiece — and the gate's whole purpose, priming a prompt cache
whose TTL is measured in minutes, cannot survive a gap that long. There is no
configuration where the pair does what its author intended, so it's an error
rather than a footgun. Use `gate: "parallel"`.

### One definition, either transport

Hardcoding `"transport" => "batch"` in a definition fixes its latency class for
every run. Often that isn't what you want: the same analysis should answer an
analyst in minutes when they click *run*, and cost half as much overnight when
a scheduler runs it across a thousand inputs. Compile with a per-run default
instead:

```elixir
# Interactive: no override, every node defaults live.
Compiler.compile(definition, input: input)

# Bulk overnight: same definition, batched, with deadline headroom for
# whatever queueing sits between the step and the provider.
Compiler.compile(definition, input: input, transport: "batch", batch_deadline: 93_600)
```

The `:transport` option fills the default for every `llm` node that doesn't
declare one; a node's own explicit `"transport"` always wins, which is how a
flow pins a cheap synthesis step live even in a batch run. `:poll_interval` and
`:batch_deadline` fill tuning defaults the same way, only on nodes that end up
batched — tuning belongs to the run, not the definition, because only the
caller knows what sits between submission and the provider.

The override merges **before validation**, so it cannot smuggle batch past the
topology rules: compiling a `sequential` fan-out with `transport: "batch"`
fails with `:sequential_fan_out` right there, while the same definition
compiles live untouched. And both the flow snapshot and each job's args carry
the merged config — what ran is what is recorded.

## Cost tracking

Batch usage flows into `workflow_step_stats` like any other call, with one
addition: the usage map carries `service_tier: "batch"` so your pricing module
can apply the discount. Nothing downstream can recover that on its own — a
stored cost looks identical either way — so branch on it:

```elixir
defmodule MyApp.Pricing do
  @behaviour Baton.Pricing

  @impl true
  def cost(%{service_tier: "batch"} = usage) do
    usage |> standard_cost() |> Decimal.mult(Decimal.new("0.5"))
  end

  def cost(usage), do: standard_cost(usage)
end
```

Live calls carry no `:service_tier` key at all.

`latency_ms` for a batch step is the **end-to-end turnaround** — from
submission to result, not the duration of any one HTTP call — which is the
number worth comparing against a live twin when you're deciding what else to
move over.

## Watching a step that waits for hours

A batch step broadcasts an `awaiting` event on submit and on every poll,
carrying the batch id and the poll interval it parked itself for:

```elixir
def handle_info({:workflow_step_updated, %{state: "awaiting"} = payload}, socket) do
  # payload.detail == %{batch_id: "msgbatch_01ABC...", seconds: 300}
  {:noreply, mark_waiting(socket, payload)}
end
```

This is distinct from `snoozed`, which means a step is waiting for a resource
of its own — its `detail.reason` says which (`"deps"`, `"rate_budget"`,
`"provider_limit"`, `"batch_slot"`, or `"step"`). The batch wait itself is
never double-reported: an `awaiting` step does not also broadcast `snoozed`.
If you pattern-match on `state` anywhere, keep a catch-all clause — new
states get added as the engine grows ways to wait.

## Rate-limiting submissions

Provider batch traffic draws from a separate pool from live traffic, so
`Baton.RateLimiter`'s `acquire/3` (which reserves input/output token budget)
doesn't apply. What can still be exceeded is the rate of submissions. Implement
the optional `acquire_batch/1` if you need to pace them:

```elixir
@impl true
def acquire_batch(opts) do
  if slots_available?(), do: {:ok, opts}, else: {:snooze, 120}
end
```

Polling is deliberately ungated — it's cheap, and asking a limiter about it
would only add latency to a step that is already waiting.

## When not to use it

- **Anything a user is waiting on.** Hours is hours.
- **Chained batch steps.** Two in a row can take 2× the latency; the DAG
  doesn't stop you, but prefer batch mode for tails or where the downstream
  work is cheap and local.
- **Steps that already coalesce.** If you have a pipeline that batches many
  items into one provider request itself, it's already getting the discount at
  better packing than one-request-per-step.

## Requirements

Batch mode needs **schema v7** for the `workflow_nodes.checkpoint` column,
which is where a step keeps its batch id between attempts:

```elixir
defmodule MyApp.Repo.Migrations.UpgradeBatonV7 do
  use Ecto.Migration
  def up,   do: Baton.Migration.up()
  def down, do: Baton.Migration.down(version: 6)
end
```

The column is only written by steps that use it; nothing else changes.