httpstream

type CompletionStreamer ¶

type CompletionStreamer interface {
	CompleteStream(ctx context.Context, req *provider.CompletionRequest) (provider.Stream, error)
}

CompletionStreamer is the minimal interface WSHandler needs from the nexus gateway. Engine.CompleteStream satisfies this implicitly.

type EventType ¶

type EventType string

EventType discriminates the wire-level event kind.

EventType is the on-the-wire form; provider.EventKind is the in-process form. Encoders translate between them. Some types (Done, Heartbeat) have no provider counterpart — they're synthesised by the runner.

const (
	EventTypeDelta     EventType = "delta"
	EventTypeReasoning EventType = "reasoning"
	EventTypeToolCall  EventType = "tool_call"
	EventTypeAudio     EventType = "audio"
	EventTypeImage     EventType = "image"
	EventTypeCitation  EventType = "citation"
	EventTypeUsage     EventType = "usage"
	EventTypeError     EventType = "error"
	EventTypeHeartbeat EventType = "heartbeat"
	EventTypeDone      EventType = "done"
)

type NDJSONEncoder ¶

type NDJSONEncoder struct{}

NDJSONEncoder emits one JSON object per line, native to the StreamEvent shape (not OpenAI-compatible). Each line is a discriminated event with a `type` field. Useful for non-browser consumers (curl -N, Go HTTP clients).

Example output:

{"type":"delta","model":"gpt-4o","delta":{"content":"Hello"}}
{"type":"reasoning","delta":{"reasoning":"Let me think..."}}
{"type":"tool_call","delta":{"tool_calls":[{"index":0,...}]}}
{"type":"usage","usage":{"prompt_tokens":5,...}}
{"type":"done"}

func NewNDJSONEncoder() *NDJSONEncoder

func (e *NDJSONEncoder) ContentType() string

func (e *NDJSONEncoder) EncodeError(w io.Writer, werr *WireError) error

func (e *NDJSONEncoder) EncodeEvent(w io.Writer, ev *StreamEvent) error

func (e *NDJSONEncoder) End(w io.Writer) error

func (e *NDJSONEncoder) Heartbeat(w io.Writer) error

func (e *NDJSONEncoder) WriteHeaders(w http.ResponseWriter)

type Registry ¶

type Registry struct {
	// contains filtered or unexported fields
}

Registry maps content types to encoders, plus aliases for convenience (e.g. "sse" → "text/event-stream"). Safe for concurrent registration during construction; treat as read-only after first request handled.

func DefaultRegistry() *Registry

func NewRegistry() *Registry

func (r *Registry) Default() StreamEncoder

func (r *Registry) Lookup(name string) StreamEncoder

func (r *Registry) Register(contentType string, enc StreamEncoder)

func (r *Registry) RegisterAlias(alias, contentType string)

func (r *Registry) SetDefault(contentType string)

type RunOptions ¶

type RunOptions struct {
	// HeartbeatInterval controls keepalive emission. The runner only sends
	// a heartbeat if no real event was flushed within the interval.
	// Zero defaults to 15s. Set to a negative value to disable.
	HeartbeatInterval time.Duration

	// RequestID is echoed into wire events for client-side correlation.
	RequestID string

	// OnError is called once when the stream terminates with an error
	// (after the encoder has emitted the typed error event but before
	// End is called). Useful for recording mid-stream failures upstream.
	// A failed write to the client is passed wrapped in ErrClientWrite;
	// use ClientGone to tell it from a provider failure.
	OnError func(error)
}

RunOptions tunes the streaming event loop. Zero value is reasonable production defaults: 15-second heartbeat, no per-event timeout, accept the request id from the caller.

type SSENativeEncoder ¶

type SSENativeEncoder struct{}

SSENativeEncoder emits events in nexus-native SSE form: each event uses a named SSE event (event: delta, event: reasoning, event: tool_call, etc.) and the data line is the raw StreamEvent JSON. This is more expressive than the OpenAI-compatible envelope — clients can route on event name without parsing the payload — but it's not compatible with OpenAI SDKs.

Use this when you control the consumer (a Go SDK, a custom dashboard). Consumers in OpenAI-SDK ecosystems should keep the SSEOpenAIEncoder.

func NewSSENativeEncoder() *SSENativeEncoder

func (e *SSENativeEncoder) ContentType() string

func (e *SSENativeEncoder) EncodeError(w io.Writer, werr *WireError) error

func (e *SSENativeEncoder) EncodeEvent(w io.Writer, ev *StreamEvent) error

func (e *SSENativeEncoder) End(w io.Writer) error

func (e *SSENativeEncoder) Heartbeat(w io.Writer) error

func (e *SSENativeEncoder) WriteHeaders(w http.ResponseWriter)

type SSEOpenAIEncoder ¶

type SSEOpenAIEncoder struct{}

SSEOpenAIEncoder emits events in the OpenAI chat.completion.chunk envelope over Server-Sent Events. The intent is drop-in compatibility with OpenAI SDKs (openai-python, openai-go, openai-node, etc.).

Tool-call deltas are forwarded with their `index`/`function.name`/ `function.arguments` fields preserved. Reasoning is surfaced as `delta.reasoning_content` (Anthropic-on-OpenAI extension already used by DeepSeek and other gateways) so OpenAI clients silently ignore it while nexus-aware clients can read it.

Final usage chunk follows OpenAI's stream_options.include_usage shape:

data: {"id":"…","object":"chat.completion.chunk","choices":[],"usage":{...}}
data: [DONE]

func NewSSEOpenAIEncoder() *SSEOpenAIEncoder

func (e *SSEOpenAIEncoder) ContentType() string

func (e *SSEOpenAIEncoder) EncodeError(w io.Writer, werr *WireError) error

func (e *SSEOpenAIEncoder) EncodeEvent(w io.Writer, ev *StreamEvent) error

func (e *SSEOpenAIEncoder) End(w io.Writer) error

func (e *SSEOpenAIEncoder) Heartbeat(w io.Writer) error

func (e *SSEOpenAIEncoder) WriteHeaders(w http.ResponseWriter)

type StreamEncoder ¶

type StreamEncoder interface {
	// ContentType is the canonical Content-Type produced by this encoder.
	ContentType() string

	// WriteHeaders writes status + headers and prepares the response for
	// streaming. Implementations should set Cache-Control: no-cache, the
	// Content-Type, and disable buffering by flushing the headers.
	WriteHeaders(w http.ResponseWriter)

	// EncodeEvent writes one event. Errors abort the stream.
	EncodeEvent(w io.Writer, e *StreamEvent) error

	// EncodeError writes a typed error event. Encoders that support named
	// events (SSE) emit a discriminated frame; NDJSON/WS just emit a
	// type:"error" object.
	EncodeError(w io.Writer, err *WireError) error

	// Heartbeat writes a keepalive frame (SSE comment, NDJSON heartbeat
	// line, or WS ping). May be a no-op for protocols with native pings.
	Heartbeat(w io.Writer) error

	// End writes any required terminator (SSE [DONE], NDJSON {"type":"done"})
	// and flushes. Does not close the underlying connection.
	End(w io.Writer) error
}

StreamEncoder writes StreamEvents to an HTTP response in a wire format.

All methods may be called from a single goroutine — encoders need not be concurrency-safe. WriteHeaders runs once before the first event; End runs once after the last (or after an error event, before the connection closes). Heartbeat is best-effort: if writing fails, the runner gives up and lets the next data write surface the error.

func Negotiate(r *http.Request, reg *Registry) StreamEncoder

type StreamEvent ¶

type StreamEvent struct {
	Type         EventType            `json:"type"`
	ID           string               `json:"id,omitempty"`
	Model        string               `json:"model,omitempty"`
	RequestID    string               `json:"request_id,omitempty"`
	Delta        *provider.Delta      `json:"delta,omitempty"`
	Usage        *provider.Usage      `json:"usage,omitempty"`
	Citation     *provider.Citation   `json:"citation,omitempty"`
	Audio        *provider.AudioChunk `json:"audio,omitempty"`
	Image        *provider.ImageChunk `json:"image,omitempty"`
	FinishReason string               `json:"finish_reason,omitempty"`
	Err          *WireError           `json:"error,omitempty"`
}

StreamEvent is the encoder-facing representation of a chunk.

Encoders should populate only the fields relevant to Type. Unset fields are omitted on the wire. ID/Model/RequestID are echoed when known.

func FromChunk(c *provider.StreamChunk, requestID string) *StreamEvent

type WSHandler ¶

type WSHandler struct {
	// contains filtered or unexported fields
}

WSHandler upgrades an HTTP request to a WebSocket connection that streams nexus events as JSON text frames. The protocol is:

client → server (text JSON):
  {"type":"start","request":{...CompletionRequest...}}
  {"type":"abort"}
  {"type":"audio_chunk","format":"pcm16","b64":"..."}    (multi-modal in)
  {"type":"image","mime_type":"image/png","b64":"..."}   (multi-modal in)

server → client (text JSON):
  StreamEvent objects (same shape as NDJSON)
  Final {"type":"done"} then graceful close.
  {"type":"error", error:{...}} on failure then close StatusInternalError,
  or StatusPolicyViolation when the gateway refused the request.

The handler is constructed with a CompletionStreamer — typically the gateway engine — that produces the provider.Stream for a parsed request. Auth lives in the parent middleware chain; the handler reads from r.Context() and respects auth/tenant decisions made before the upgrade.

func NewWSHandler(streamer CompletionStreamer, opts WSOptions) *WSHandler

func (h *WSHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)

type WSOptions ¶

type WSOptions struct {
	// AcceptOrigins is the allowed list for the Origin check. Empty means
	// the same-origin policy (nhooyr default). Pass `[]string{"*"}` to
	// allow any origin (use only for trusted networks).
	AcceptOrigins []string

	// SendBuffer is the buffered channel size feeding the writer goroutine.
	// Default: 64.
	SendBuffer int

	// HeartbeatInterval, when > 0, sends WS pings on the configured cadence
	// to keep the connection alive across idle proxies. Default: 20s.
	HeartbeatInterval time.Duration

	// OnError is called once when a stream fails, with the error as the
	// stream returned it. The client sees only the sanitized envelope, so
	// this is where the cause is logged. ctx carries the request id.
	OnError func(ctx context.Context, err error)
}

WSOptions configures the WebSocket handler.

type WireError ¶

type WireError struct {
	Message   string `json:"message"`
	Type      string `json:"type"`
	Code      string `json:"code,omitempty"`
	Retryable bool   `json:"retryable"`
	RequestID string `json:"request_id,omitempty"`
}

WireError is the sanitised error envelope. Internal error strings are intentionally not exposed; consumers see a Type, a request id, and a retryable flag.

func SanitizeError(err error, requestID string) *WireError