# `LangExtract.Runner.Delivery`
[🔗](https://github.com/mdepolli/lang_extract/blob/v0.10.0/lib/lang_extract/runner/delivery.ex#L1)

Bounded delivery of supervised chunk results to a lazy caller-side stream.

The consumer's demand drives admission: at most `:buffer` tasks are
outstanding, so the caller's mailbox never holds more than `:buffer`
undelivered results (plus their `:DOWN` notices). A slow consumer
throttles admission instead of growing a mailbox — the normative
bounded-delivery constraint from the production-pipeline design.

Task crashes become per-chunk events via the monitor `:DOWN` reason;
halting the stream early kills all outstanding tasks.

## Why hand-rolled

Without drain, this module should collapse into
`Task.Supervisor.async_stream_nolink(..., max_concurrency: buffer,
ordered: false, zip_input_on_exit: true)` — the bounded/unordered/
crash-to-event core duplicates it. Drain is what it can't express:
reporting never-started chunks requires detecting the dead supervisor
at admission time. Layering that detection on top was also considered
and rejected: a `Stream.transform` can't catch the supervisor-down
exit (it erupts between elements, inside the inner enumerable's
reduction), so the wrapper needs continuation-driving via
`Enumerable.reduce` plus OTP exit-shape matching plus reconstructing
the never-started set by diffing observed byte ranges — more fragile
than this explicit loop, and dependent on uncontracted stdlib event
ordering. If drain semantics are ever dropped, delete this module in
favor of `async_stream_nolink`.

## Drain

When the runner shuts down mid-stream (deploy, supervisor restart),
chunk tasks trap exits, so the supervisor's shutdown signal gives each
in-flight request up to `:shutdown` milliseconds to finish naturally —
completed results still reach the consumer. Chunks never started are
emitted as `{:error, %ChunkError{reason: :drained}}`, detected when
admission hits the dead task supervisor. Consumers already handle
per-chunk errors, so drain adds no new consumer code paths.

Internal — no stability guarantees; see the README's "Stability"
section. Documented because it explains how the library works, not
because it is API.

# `event`

```elixir
@type event() ::
  {:ok, LangExtract.ChunkResult.t()} | {:error, LangExtract.ChunkError.t()}
```

# `stream_events`

```elixir
@spec stream_events(
  Supervisor.supervisor(),
  [LangExtract.Chunker.Chunk.t()],
  (LangExtract.Chunker.Chunk.t() ->
     {:ok, LangExtract.ChunkResult.t()} | {:error, LangExtract.ChunkError.t()}),
  keyword()
) :: Enumerable.t()
```

---

*Consult [api-reference.md](api-reference.md) for complete listing*