Provider-neutral, bounded, resumable execution for durable LM batches.
Requests have stable caller-supplied IDs and JSON-safe payloads. A dispatcher receives each normalized request plus attempt metadata and must return one of:
{:ok, output}{:transient, reason}- the request did not run and may be sent again{:terminal, reason}{:malformed, reason}{:ambiguous, reason}- the request may have run; it is not sent again
A request that may already have run is never re-dispatched: sending it again
could run and bill it twice. Only a :transient outcome is retried, up to
:max_attempts. A dispatcher that raises, throws or exits, or that is still
running at :timeout, may have sent its request first, so its outcome is
:ambiguous.
The batch owns the wait before a retry as well as the retry. When the
failure carries the provider's retry-after header (seconds or an HTTP
date), the request is not sent again before then; otherwise the wait grows
exponentially with the attempt, with jitter. A request that is waiting holds
no dispatch slot: other requests are sent meanwhile, and the batch sleeps
only when every remaining request is waiting. A process that traps exits
and is stopped by its parent during that sleep exits at once with the
parent's reason; a dispatch wave that is running is still waited for, for
up to :timeout. The parent is the first of the process's $ancestors,
which a process started by Task, GenServer or another OTP behaviour
has; a process started with bare spawn/1 or spawn_link/1 has none, and
its sleep cannot be interrupted. Backoff is capped at :max_retry_wait.
When the provider asks for longer than :max_retry_wait,
or the wait would end after the Imp.Deadline bound to the calling
process, the request is not retried in this run: it stays
:transient_failure with the attempts it has used, the summary is not
complete?, and resume/3 retries it. The time a request may be sent
again is kept in the checkpoint as a UTC time, and resume/3 waits for it
under the same rules.
The checkpoint is an auditable JSON document. Every dispatch intent and
outcome is appended to its event history before execution continues. Writes
use a synced temporary file followed by an atomic rename. On resume, a
request that was dispatched but has no committed outcome becomes
:ambiguous for the same reason.
A checkpoint written by Imp 0.5.0 (schema version 1) retried timeouts and
dispatcher crashes as transient failures, and its stored reason does not
always say which a failure was. On resume, its :transient_failure
requests become :ambiguous and are not sent again; the checkpoint is
rewritten at schema version 2.
Summary
Types
@type context() :: %{request_id: String.t(), attempt: pos_integer()}
@type summary() :: %{ checkpoint: Path.t(), complete?: boolean(), counts: %{required(atom()) => non_neg_integer()}, requests: [map()] }
Functions
@spec req_llm_dispatcher(Imp.Clients.ReqLLM.t(), keyword()) :: dispatcher()
Builds a dispatcher backed by an existing ReqLLM client.
Request payloads may be a message list or %{"messages" => messages}. The
adapter is provider-neutral because provider and model selection remain in
the client ("openai:...", "anthropic:...", or "gemini:...").
The batch owns retries: every call is made with max_retries: 0, so ReqLLM
sends each attempt once and the error describes the only send.
A failed call's Imp.LMError decides its outcome:
:transient- the request never reached the provider (the connection was refused, or no pooled connection was free), or the provider answered that it did not process it and to try later (408, 425, 429, 503, 529).:terminal- the provider rejected it with any other 4xx status, or the call failed with no status and no transport reason.:ambiguous- it may have run: a 5xx status other than 503 or 529 (500 included), any other status, or a timeout or closed connection with no response.
@spec resume(Path.t(), dispatcher(), keyword()) :: {:ok, summary()} | {:error, term()}
Resumes a batch from its checkpoint.
Persisted requests, attempts, and retry policy are authoritative. Resume
accepts only runtime options: :num_threads, :timeout,
:validate_output, :max_retry_wait and :clock. A request waiting for
a retry, or whose retry was stopped, is not sent before the not_before
time its checkpoint holds: the batch waits for it when that is within
:max_retry_wait and the deadline, and stops its retry again, unsent,
when it is not.
@spec run([map()], dispatcher(), keyword()) :: {:ok, summary()} | {:error, term()}
Starts a new durable batch.
Required options:
:checkpoint- destination for the atomic JSON checkpoint
Runtime options:
:num_threads- requests dispatched at once (default4):max_attempts- sends per request, retries included (default3):timeout- milliseconds one dispatch may run (default30_000):validate_output- an optional arity-one callback returning:okor{:error, reason}:max_retry_wait- the longest wait, in milliseconds, before a retry (default60_000):clock- the time source and sleep the retry wait uses, a map of:now(monotonic milliseconds),:utc_now(aDateTime) and:sleep(milliseconds) functions; it exists so tests need not wait