Add WebSocket endpoint for streaming audio transcription

README.md

@@ -12,6 +12,7 @@ A unified speech-to-text and translation service that provides a single API for
 
 - **Command Line Interface**: Downloadable CLI for communicating with the server for  audio processing
 - **Download and Realtime Transcription** with JSON, SRT, VTT, and text output formats
+- **Streaming Audio Input**: Real-time transcription of audio streams via WebSocket
 - **HTTP API Server**: RESTful API for transcription and translation service
 - **Docker Support**: Pre-built GPU-enabled containers for easy deployment of the service
 - **GPU Support**: CUDA, Vulkan, and Metal (macOS) acceleration for local models
@@ -170,8 +171,8 @@ flowchart LR
         end
     end
 
-    CLI --> API
-    SDK --> API
+    CLI -->|REST / WebSocket| API
+    SDK -->|REST / WebSocket| API
     API --> Orch
     Orch --> WS
     WS --> Models

doc/API.md

@@ -168,6 +168,75 @@ Translates audio files to English text.
 
 **Response Formats:** Same as transcription (JSON, plain text, SRT, VTT)
 
+### Streaming Transcription (WebSocket)
+
+```http
+GET /api/whisper/stream
+```
+
+Upgrades the connection to a WebSocket for real-time streaming audio transcription. The client sends raw audio data and receives transcription segments as they are produced, using a sliding-window approach internally.
+
+**Protocol:**
+
+1. Connect via WebSocket (`ws://` or `wss://`)
+2. Send a JSON configuration message as the first text frame
+3. Receive a `stream.ready` event
+4. Send binary frames containing 16-bit signed little-endian PCM audio at 16 kHz mono
+5. Receive `stream.segment` events as transcription segments are produced
+6. Close the connection when done; remaining audio is flushed and a `stream.done` event is sent
+
+**Configuration Message (first text frame):**
+
+```json
+{
+  "model": "ggml-medium-q5_0",
+  "language": "en",
+  "temperature": 0.0,
+  "prompt": "Optional initial prompt",
+  "translate": false,
+  "step_ms": 3000,
+  "length_ms": 10000,
+  "keep_ms": 200,
+  "vad_threshold": 0.0
+}
+```
+
+| Field | Type | Default | Description |
+|-------|------|---------|-------------|
+| `model` | string | (required) | Model ID to use (must be a local whisper model) |
+| `language` | string | `"auto"` | Language code |
+| `temperature` | number | `0.0` | Sampling temperature (0–1) |
+| `prompt` | string | | Initial prompt to guide transcription |
+| `translate` | boolean | `false` | Translate to English |
+| `step_ms` | integer | `3000` | How often to run inference on new audio (ms) |
+| `length_ms` | integer | `10000` | Maximum window size fed to the model (ms) |
+| `keep_ms` | integer | `200` | Overlap retained between consecutive windows (ms) |
+| `vad_threshold` | number | `0.0` | RMS energy threshold for voice activity detection (0 = disabled) |
+
+**Server Events (text frames):**
+
+```json
+{"type": "stream.ready"}
+```
+
+```json
+{"type": "stream.segment", "segment": {"id": 0, "start": 0.0, "end": 2.5, "text": "Hello world."}}
+```
+
+```json
+{"type": "stream.error", "error": "error message"}
+```
+
+```json
+{"type": "stream.done"}
+```
+
+**Notes:**
+
+- Only local whisper models are supported for streaming; cloud providers (OpenAI, ElevenLabs) are not available via this endpoint.
+- Audio must be 16 kHz, mono, 16-bit signed little-endian PCM. No resampling is performed server-side.
+- Each streaming session holds a model context for its duration. With a limited pool size (`--whisper.max-contexts`), long-running streams may prevent other requests from being served.
+
 ## Error Handling
 
 The API uses standard HTTP status codes and returns JSON error responses:

go.mod

@@ -6,6 +6,7 @@ toolchain go1.24.2
 
 require (
 	github.com/alecthomas/kong v1.13.0
+	github.com/coder/websocket v1.8.14
 	github.com/djthorpe/go-errors v1.0.3
 	github.com/go-audio/wav v1.1.0
 	github.com/mutablelogic/go-client v1.3.5

go.sum

@@ -20,6 +20,8 @@ github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1x
 github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=
+github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
 github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI=
 github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M=
 github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE=

pkg/httphandler/httphandler.go

@@ -24,6 +24,7 @@ func RegisterHandlers(router *http.ServeMux, prefix string, manager *pkg.Manager
 	RegisterModelHandlers(router, prefix, manager, middleware)
 	RegisterTranscribeHandlers(router, prefix, manager, middleware)
 	RegisterTranslateHandlers(router, prefix, manager, middleware)
+	RegisterStreamHandlers(router, prefix, manager, middleware)
 }
 
 ///////////////////////////////////////////////////////////////////////////////

pkg/httphandler/stream.go

@@ -0,0 +1,124 @@
+package httphandler
+
+import (
+	"context"
+	"encoding/binary"
+	"net/http"
+
+	// Packages
+	httpresponse "github.com/mutablelogic/go-server/pkg/httpresponse"
+	pkg "github.com/mutablelogic/go-whisper/pkg"
+	schema "github.com/mutablelogic/go-whisper/pkg/schema"
+	websocket "github.com/coder/websocket"
+	wsjson "github.com/coder/websocket/wsjson"
+)
+
+///////////////////////////////////////////////////////////////////////////////
+// PUBLIC METHODS
+
+// RegisterStreamHandlers registers the WebSocket handler for streaming audio transcription
+func RegisterStreamHandlers(router *http.ServeMux, prefix string, manager *pkg.Manager, middleware HTTPMiddlewareFuncs) {
+	// Do NOT wrap with middleware — it may interfere with the WebSocket upgrade
+	router.HandleFunc(joinPath(prefix, "stream"), func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodGet {
+			_ = httpresponse.Error(w, httpresponse.Err(http.StatusMethodNotAllowed), r.Method)
+			return
+		}
+		_ = streamHandle(w, r, manager)
+	})
+}
+
+///////////////////////////////////////////////////////////////////////////////
+// PRIVATE METHODS
+
+// streamHandle upgrades the connection to WebSocket and runs the streaming loop
+func streamHandle(w http.ResponseWriter, r *http.Request, manager *pkg.Manager) error {
+	// Upgrade to WebSocket
+	conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{})
+	if err != nil {
+		return err
+	}
+
+	// Use a cancellable context derived from the request context
+	ctx, cancel := context.WithCancel(r.Context())
+	defer cancel()
+
+	// Read first text message as StreamConfig JSON
+	var cfg schema.StreamConfig
+	if err := wsjson.Read(ctx, conn, &cfg); err != nil {
+		conn.Close(websocket.StatusUnsupportedData, "expected StreamConfig JSON as first message")
+		return err
+	}
+
+	// Create a new stream session from the manager
+	session, cleanup, err := manager.NewStreamSession(ctx, &cfg)
+	if err != nil {
+		_ = wsjson.Write(ctx, conn, schema.StreamEvent{Type: schema.StreamEventError, Error: err.Error()})
+		conn.Close(websocket.StatusInternalError, err.Error())
+		return err
+	}
+	defer cleanup()
+
+	// Notify client that we are ready
+	if err := wsjson.Write(ctx, conn, schema.StreamEvent{Type: schema.StreamEventReady}); err != nil {
+		conn.Close(websocket.StatusInternalError, err.Error())
+		return err
+	}
+
+	// Main loop: read binary PCM frames from client
+	for {
+		msgType, data, err := conn.Read(ctx)
+		if err != nil {
+			// Client closed or error — run final processing
+			break
+		}
+
+		if msgType != websocket.MessageBinary {
+			// Unexpected message type — skip
+			continue
+		}
+
+		// Convert 16-bit signed little-endian PCM to float32
+		samples := make([]float32, len(data)/2)
+		for i := range samples {
+			raw := binary.LittleEndian.Uint16(data[i*2 : i*2+2])
+			samples[i] = float32(int16(raw)) / 32768.0
+		}
+
+		session.AddSamples(samples)
+
+		if session.ShouldProcess() {
+			segments, err := session.Process(ctx)
+			if err != nil {
+				_ = wsjson.Write(ctx, conn, schema.StreamEvent{Type: schema.StreamEventError, Error: err.Error()})
+				conn.Close(websocket.StatusInternalError, err.Error())
+				return err
+			}
+			for _, seg := range segments {
+				if err := wsjson.Write(ctx, conn, schema.StreamEvent{Type: schema.StreamEventSegment, Segment: seg}); err != nil {
+					conn.Close(websocket.StatusInternalError, err.Error())
+					return err
+				}
+			}
+		}
+	}
+
+	// Process any remaining audio
+	segments, err := session.ProcessFinal(ctx)
+	if err != nil {
+		_ = wsjson.Write(ctx, conn, schema.StreamEvent{Type: schema.StreamEventError, Error: err.Error()})
+		conn.Close(websocket.StatusInternalError, err.Error())
+		return err
+	}
+	for _, seg := range segments {
+		if err := wsjson.Write(ctx, conn, schema.StreamEvent{Type: schema.StreamEventSegment, Segment: seg}); err != nil {
+			conn.Close(websocket.StatusInternalError, err.Error())
+			return err
+		}
+	}
+
+	// Send done event and close cleanly
+	_ = wsjson.Write(ctx, conn, schema.StreamEvent{Type: schema.StreamEventDone})
+	conn.Close(websocket.StatusNormalClosure, "")
+	return nil
+}

pkg/manager.go

@@ -268,6 +268,56 @@ func (m *Manager) Translate(ctx context.Context, w schema.SegmentWriter, r io.Re
 	}
 }
 
+// NewStreamSession creates a new streaming transcription session for the
+// specified model. The caller MUST call the returned release function when
+// the session is finished to return the task to the pool.
+// Streaming is only supported for local whisper models.
+func (m *Manager) NewStreamSession(ctx context.Context, cfg *schema.StreamConfig) (*whisper.StreamSession, func(), error) {
+	ctx, endSpan := otel.StartSpan(m.tracer, ctx, "manager.NewStreamSession",
+		attribute.String("model", cfg.Model),
+	)
+	var err error
+	defer func() { endSpan(err) }()
+
+	// Resolve the model
+	model, err := m.GetModel(ctx, cfg.Model)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Streaming is only supported for local whisper models
+	if model.OwnedBy != "whisper" {
+		err = httpresponse.ErrBadRequest.With("streaming is only supported for local whisper models")
+		return nil, nil, err
+	}
+
+	// Acquire a long-lived task from the pool
+	task, release, err := m.whisper.AcquireTask(model)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Configure task parameters from the stream config
+	if cfg.Language != nil {
+		task.SetLanguage(*cfg.Language)
+	}
+	if cfg.Temperature != nil {
+		if err = task.SetTemperature(*cfg.Temperature); err != nil {
+			release()
+			return nil, nil, err
+		}
+	}
+	if cfg.Prompt != nil {
+		task.SetPrompt(*cfg.Prompt)
+	}
+	if cfg.Translate {
+		task.SetTranslate(true)
+	}
+
+	session := whisper.NewStreamSession(task, *cfg)
+	return session, release, nil
+}
+
 ///////////////////////////////////////////////////////////////////////////////
 // PRIVATE METHODS - WHISPER
 

pkg/schema/stream.go

@@ -0,0 +1,47 @@
+package schema
+
+//////////////////////////////////////////////////////////////////////////////
+// TYPES
+
+// StreamConfig is the configuration message sent by the client as the first
+// WebSocket text message when initiating a streaming transcription session.
+type StreamConfig struct {
+	Model        string   `json:"model"`
+	Language     *string  `json:"language,omitempty"`
+	Temperature  *float64 `json:"temperature,omitempty"`
+	Prompt       *string  `json:"prompt,omitempty"`
+	Translate    bool     `json:"translate,omitempty"`
+	StepMs       int      `json:"step_ms,omitempty"`
+	LengthMs     int      `json:"length_ms,omitempty"`
+	KeepMs       int      `json:"keep_ms,omitempty"`
+	VadThreshold float32  `json:"vad_threshold,omitempty"`
+}
+
+// StreamEvent is a JSON message sent back to the client during a streaming
+// transcription session.
+type StreamEvent struct {
+	Type    string   `json:"type"`
+	Segment *Segment `json:"segment,omitempty"`
+	Error   string   `json:"error,omitempty"`
+}
+
+//////////////////////////////////////////////////////////////////////////////
+// GLOBALS
+
+const (
+	StreamEventReady   = "stream.ready"
+	StreamEventSegment = "stream.segment"
+	StreamEventError   = "stream.error"
+	StreamEventDone    = "stream.done"
+)
+
+//////////////////////////////////////////////////////////////////////////////
+// STRINGIFY
+
+func (c *StreamConfig) String() string {
+	return stringify(*c)
+}
+
+func (e *StreamEvent) String() string {
+	return stringify(*e)
+}