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 (ALLM.SpeechStreamAdapter.stream_synthesize_input/3, 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.