Spectre.Inference.StreamAdapter behaviour (Spectre v0.3.3)

Copy Markdown View Source

Provider-neutral streaming contract owned by the Spectre core.

Pull adapters should request at most one transport item per request_transport_item/1 call. Push adapters must declare both :push_transport and :bounded_push_transport: the adapter owns the bound between its producer and the session mailbox, while Spectre independently bounds normalized events inside the session. Declaring only :push_transport is rejected because an Erlang mailbox is not itself a backpressure mechanism.

Normalized usage counters are cumulative for the logical attempt, including after resume/3. Spectre passes the durable usage floor in :resume_usage; adapters for providers that restart counters locally must add that floor before emitting ProviderEvent values.

An adapter declaring :cost_usage promises that the cumulative cost counter is authoritative under the immutable pricing reference configured for the invocation. Spectre rejects hard cost budgets when this capability is absent.

Callbacks run in the session process and therefore must return promptly. Adapters should use asynchronous transport messages instead of blocking open/2, resume/3, request_transport_item/1, or cancel/2 indefinitely. Orderly shutdown gives cancel/2 a one-second best-effort window before the session is killed.

Provider data is always delivered through the session mailbox. open/2 and resume/3 run in that session process, so self() is the destination an asynchronous transport helper must retain. Pull and push capabilities describe demand, not an alternative delivery channel.

The session traps exits so its termination callback can cancel an open provider request. Consequently, a provider helper linked to the session delivers {:EXIT, pid, reason} through handle_transport/2 instead of killing the session. Adapters should prefer monitors when helper death is a protocol failure, or explicitly return {:ignore, state} for unrelated or expected exit messages. An ignored helper crash is eventually classified by the provider-stall timeout if no owned transport item arrives.

handle_transport/2 must return {:ignore, state} only for a mailbox message that did not consume the outstanding pull item. A consumed transport item that yields no logical events returns {:ok, [], state} so Spectre can request the next item without allowing two requests in flight.

Spectre passes positive adapter-owned limits under the :spectre_bounds keyword option to open/2 and resume/3. The values below are defaults, not constants an adapter should copy:

spectre_bounds: [
  max_transport_chunk_bytes: 256_000,
  max_parser_residual_bytes: 256_000
]

Every adapter must enforce both limits before retaining bytes and fail with :provider_stream_overflow; it must never truncate or drop provider text. Applications can override the resolved values with :stream_max_transport_chunk_bytes and :stream_max_parser_residual_bytes on an inference. The conformance runner requires conformance_fixture/4. It supplies a deterministic binary larger than the selected bound; the adapter wraps that binary in its real transport-message shape and returns isolated parser state. The runner then invokes handle_transport/2 and requires the overflow reply.

Delta payloads are binary transport fragments. They need not end on a UTF-8 codepoint boundary; Spectre incrementally reassembles and validates UTF-8 before yielding public StreamEvent values.

ProviderEvent.provider_sequence is optional only for an entirely unnumbered attempt. Once any event supplies a non-negative sequence, every later event -- including usage and terminal events -- must supply exactly the next integer. Mixing numbered deltas with unnumbered usage is rejected as :provider_sequence_violation.

Provider packages can run Spectre.Inference.StreamAdapter.Conformance against deterministic transport fixtures without depending on ExUnit.

Summary

Types

adapter_state()

@type adapter_state() :: term()

conformance_bound()

@type conformance_bound() :: :transport_chunk | :parser_residual

descriptor()

@type descriptor() :: Spectre.Inference.Descriptor.t()

provider_metadata()

@type provider_metadata() :: map()

Callbacks

cancel(adapter_state, term)

@callback cancel(adapter_state(), term()) :: :ok | {:error, term()}

capabilities(term, keyword)

@callback capabilities(
  term(),
  keyword()
) :: MapSet.t(atom())

conformance_fixture(conformance_bound, binary, descriptor, keyword)

(optional)
@callback conformance_fixture(conformance_bound(), binary(), descriptor(), keyword()) ::
  {:ok, term(), adapter_state()} | {:error, term()}

Builds an isolated, oversized fixture for the conformance runner.

The second argument is a runner-owned binary whose byte size is one greater than the configured bound. The fixture must incorporate it as the raw transport chunk or incomplete parser fragment named by the first argument. The returned message and adapter state are passed directly to handle_transport/2. Provider packages claiming stream conformance must implement both bound kinds and make that call return {:error, :provider_stream_overflow, state}. This callback is test-only and is not required by runtime capability negotiation.

handle_transport(term, adapter_state)

@callback handle_transport(term(), adapter_state()) ::
  {:ok, [Spectre.Inference.ProviderEvent.t()], adapter_state()}
  | {:ignore, adapter_state()}
  | {:error, term(), adapter_state()}

open(descriptor, keyword)

@callback open(
  descriptor(),
  keyword()
) :: {:ok, adapter_state(), provider_metadata()} | {:error, term()}

reconcile(descriptor, term, keyword)

(optional)
@callback reconcile(descriptor(), term(), keyword()) ::
  {:ok, term()} | :pending | :not_found | {:error, term()}

request_transport_item(adapter_state)

(optional)
@callback request_transport_item(adapter_state()) ::
  {:ok, adapter_state()} | {:error, term()}

resume(descriptor, term, keyword)

(optional)
@callback resume(descriptor(), term(), keyword()) ::
  {:ok, adapter_state(), provider_metadata()} | {:error, term()}

Functions

validate(adapter, profile, opts \\ [])

@spec validate(module(), term(), keyword()) ::
  {:ok, MapSet.t(atom())} | {:error, term()}