# `ALLM.Providers.Support.InputPump`
[🔗](https://github.com/cykod/ALLM/blob/v0.6.1/lib/allm/providers/support/input_pump.ex#L1)

Reduces a caller-supplied input enumerable in a helper process and hands
its elements to an owner process under a credit window.

Layer B — runtime support for the input-streaming audio callbacks
(`c:ALLM.SpeechStreamAdapter.stream_synthesize_input/3`,
`c:ALLM.TranscriptionStreamAdapter.stream_transcribe/3`).

## Why a helper process

A microphone or LLM enumerable blocks while it waits for its next
element, and the process reducing the audio stream must keep reading
server frames during that wait. So the input is reduced in a separate
process, the pump, and each element arrives in the owner's mailbox as a
message the owner can `receive` alongside its transport messages. The
owner is the only process that writes the transport.

**Consequence for callers:** the input runs in another process. An input
that reads the caller's mailbox (`Stream.repeatedly(fn -> receive … end)`)
or process dictionary sees the pump's instead, and must be relayed from a
process the caller controls.

## Messages the owner receives

    {ref, {:input, element}}
    {ref, :input_done}
    {ref, {:input_error, %{kind: kind, message: message}}}
    {:DOWN, ref, :process, pid, reason}

`ref` is the monitor reference `start/3` returns, so every pump message
and the pump's `:DOWN` share one tag. `:input_error` reports an input that
raised, threw or exited inside its own reduction; its payload is a
`%{kind, message}` map (`kind` is `:error`, `:throw` or `:exit`, `message` is
`Exception.format_banner/3`), never the raw exception or exit term, which
can carry pids and references. A `:DOWN` whose reason is not `:normal`
and that was not preceded by `:input_done` or `:input_error` means the
pump was killed by an exit signal, typically from a process linked inside
the input that crashed.

## Lifecycle

  1. `start/3` spawns the pump with `spawn_monitor/1`. The pump is **not
     linked** to the caller, so its exit is a `:DOWN` message and never
     an exit signal that could take the caller down with it.
  2. The pump links a watchdog that monitors the owner and kills the pump
     when the owner dies, so a killed consumer never leaves an input being
     reduced.
  3. At most `window` elements are unacknowledged at any time. The pump
     blocks until the owner calls `ack/2`.
  4. `stop/2` kills the pump, waits for it to be dead, and drains every
     `{ref, _}` message and the `:DOWN` from the caller's mailbox. It is
     idempotent.

A killed pump runs no code, so the input's own cleanup (a
`Stream.resource/3` after function) does not run. Its resources are
released by process exit instead: processes linked to the pump, such as
the request process `Finch.async_request/3` links to its caller, die with
it.

`start/3` must be called from the process that will receive the `:DOWN`,
which is the owner in every bundled adapter.

## Consuming pump messages

Every consumer handles the four messages the same way, so the protocol
lives here rather than in each adapter: `is_pump_message/2` selects them
in the owner's `receive` (next to any transport clauses), and `classify/2`
says what each one means. The consumer keeps only its element rules and
its own error construction:

    receive do
      message when InputPump.is_pump_message(message, ref) ->
        case InputPump.classify(message, ref) do
          {:input, element} -> InputPump.ack(pid, ref); handle(element)
          :done -> finish()
          {:failed, cause, info} -> fail(cause, info)
        end
    after
      timeout -> time_out()
    end

`default_window/0` is the credit window the bundled adapters use when the
caller does not set one.

# `classified`

```elixir
@type classified() ::
  {:input, term()}
  | :done
  | {:failed, :input_raised | :input_crashed, input_error()}
```

What a pump message means to its owner. `cause` is the value an adapter
puts on its error's `metadata.cause`; `info` goes on the error's `:cause`.

# `input_error`

```elixir
@type input_error() :: %{kind: :error | :throw | :exit, message: String.t()}
```

An input failure report: an atom `kind` and a string `message`, safe to encode and persist.

# `ack`

```elixir
@spec ack(pid(), reference()) :: :ok
```

Grant the pump one more element of credit.

# `classify`

```elixir
@spec classify(tuple(), reference()) :: classified()
```

Classify a pump message selected by `is_pump_message/2`.

  * `{ref, {:input, element}}` — `{:input, element}`. The owner must call
    `ack/2` once it has taken the element, or the pump stalls when the
    window is spent.
  * `{ref, :input_done}` — `:done`; the input is exhausted.
  * `{ref, {:input_error, info}}` — `{:failed, :input_raised, info}`; the
    input raised, threw or exited inside its own reduction.
  * `{:DOWN, ref, :process, _, reason}` — `{:failed, :input_crashed,
    crash_info(reason)}`; an exit signal killed the pump.

Pure: it neither acks nor stops the pump.

## Examples

    iex> ref = make_ref()
    iex> ALLM.Providers.Support.InputPump.classify({ref, {:input, "a"}}, ref)
    {:input, "a"}
    iex> ALLM.Providers.Support.InputPump.classify({:DOWN, ref, :process, self(), :boom}, ref)
    {:failed, :input_crashed, %{kind: :exit, message: "** (exit) :boom"}}

# `crash_info`

```elixir
@spec crash_info(term()) :: input_error()
```

Describe a pump's abnormal `:DOWN` reason as the same `%{kind, message}` map an
`:input_error` carries, with `kind: :exit`.

An adapter puts it on the error's `:cause` field for an input that
crashed, so the raw exit reason, which can carry pids and references,
never reaches the error.

## Examples

    iex> ALLM.Providers.Support.InputPump.crash_info(:boom)
    %{kind: :exit, message: "** (exit) :boom"}

# `default_window`

```elixir
@spec default_window() :: pos_integer()
```

The credit window bundled adapters use when `adapter_opts[:input_window]`
is not set.

## Examples

    iex> ALLM.Providers.Support.InputPump.default_window()
    8

# `is_pump_message`
*macro* 

Guard: `message` is one of the pump's messages for `ref` (a `{ref, _}`
tuple or the pump's `:DOWN`). Use it to select pump messages in a
selective `receive`, so no other message in the owner's mailbox is
consumed.

# `start`

```elixir
@spec start(Enumerable.t(), pid(), pos_integer()) :: {pid(), reference()}
```

Start a pump reducing `enumerable` and sending its elements to `owner`.

Returns `{pid, ref}`, where `ref` is both the monitor reference and the
tag on every message the pump sends. `window` is the number of elements
the pump may send before it waits for an `ack/2`.

# `stop`

```elixir
@spec stop(pid(), reference()) :: :ok
```

Kill the pump and remove every trace of it from the caller's mailbox.

After it returns, the pump is dead, the monitor is gone, and no
`{ref, _}` message or `:DOWN` for `ref` remains in the mailbox. Calling it
again, or after the pump already finished, is a no-op.

---

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