Streaming And SSE

Pristine.stream/3 consumes the same Pristine.Operation envelope as Pristine.execute/3. Streaming transports return a Pristine.Response whose :stream field yields events lazily.

Build A Streaming Client

This direct streaming client is standalone compatibility. Governed streaming uses governed_authority and rejects env-backed default_auth and direct headers.

client =
  Pristine.Client.new(
    base_url: "https://api.example.com",
    stream_transport: Pristine.Adapters.Transport.FinchStream,
    transport_opts: [finch: MyApp.Finch],
    serializer: Pristine.Adapters.Serializer.JSON,
    default_auth: [Pristine.Adapters.Auth.Bearer.new("example-api-token")]
  )

authority =
  Pristine.GovernedAuthority.new!(
    base_url: "https://api.example.com",
    base_url_ref: "base-url://example/workspace-123",
    credential_handle_ref: "credential-handle://example/workspace-123",
    credential_lease_ref: "credential-lease://example/one-effect",
    target_ref: "target://example/production",
    request_scope_ref: "request-scope://example/widgets/list",
    header_policy_ref: "header-policy://example/default",
    materialization_kind: "bearer",
    bearer_token_ref: "bearer-token://example/one-effect",
    redaction_ref: "redaction://headers",
    headers: %{"x-authority-target" => "target://example/production"},
    credential_headers: %{"authorization" => "Bearer authority-materialized-token"},
    allowed_header_names: ["authorization", "x-authority-target"]
  )

client =
  Pristine.Client.new(
    governed_authority: authority,
    stream_transport: Pristine.Adapters.Transport.FinchStream,
    transport_opts: [finch: MyApp.Finch],
    serializer: Pristine.Adapters.Serializer.JSON
  )

Stream An SSE Operation

operation =
  Pristine.Operation.new(%{
    id: "events.stream",
    method: :get,
    path_template: "/v1/events",
    headers: %{"accept" => "text/event-stream"},
    auth: %{
      use_client_default?: true,
      override: nil,
      security_schemes: ["bearerAuth"]
    },
    runtime: %{
      resource: "events",
      retry_group: "events.read",
      circuit_breaker: "events_api",
      rate_limit_group: "events.integration",
      telemetry_event: [:my_sdk, :events, :stream],
      timeout_ms: nil
    }
  })

{:ok, response} = Pristine.stream(client, operation)

response.stream
|> Enum.each(fn event ->
  IO.inspect(event, label: "event")
end)

For SSE helpers, combine Pristine.stream/3 with Pristine.Streaming.Event, Pristine.Streaming.SSEDecoder, and the stream transport adapters.

Unary Versus Streaming Cancellation

The new unary Pristine.Cancellation contract does not redesign the existing SSE stream lifecycle. Pristine.Adapters.Transport.FinchStream keeps its established stream cancellation/early-halt cleanup behavior and last-event-id bookkeeping. Unary capability discovery must not be interpreted as a streaming guarantee, and stream cancellation must not be inferred from :unary_cancellation.

A future common capability surface may describe streaming guarantees separately when they are backed by the existing stream acceptance tests.