fix(voice): harden streaming concurrency, provider resilience, and codec labeling

voice/audio/codec.go

@@ -10,6 +10,7 @@ const (
 	CodecMP3
 	CodecOpus
 	CodecWAV
+	CodecFLAC
 )
 
 // Format describes the encoding parameters of an audio stream.

voice/audio/resample.go

@@ -62,6 +62,15 @@ func ResamplePCM16(data []byte, fromRate, toRate, channels int) []byte {
 // ResampleStream wraps a Stream[Frame] and resamples every PCM16 frame to
 // targetRate on the fly. Non-PCM frames or frames already at the target
 // rate are passed through unchanged.
+//
+// Leak note: the worker goroutine blocks in input.Read(). Read is a
+// blocking, non-interruptible call, so if the consumer interrupts the
+// returned stream while the worker is parked inside input.Read() on a
+// stalled upstream, the worker cannot exit until that Read returns. The
+// upstream input MUST therefore eventually become readable or return an
+// error (e.g. by being closed/interrupted) for the worker to terminate.
+// As a cheap safeguard the worker checks for an output interrupt before
+// each Read so it exits promptly when interrupted between frames.
 func ResampleStream(input Stream[Frame], targetRate int) Stream[Frame] {
 	if targetRate <= 0 {
 		return input
@@ -70,6 +79,13 @@ func ResampleStream(input Stream[Frame], targetRate int) Stream[Frame] {
 	go func() {
 		defer p.Close()
 		for {
+			// Exit early if the consumer already interrupted the output,
+			// avoiding a blocking Read we would immediately discard.
+			select {
+			case <-p.ctx.Done():
+				return
+			default:
+			}
 			f, err := input.Read()
 			if err != nil {
 				if err != io.EOF {

voice/audio/stream.go

@@ -53,6 +53,24 @@ func (p *Pipe[T]) Read() (T, error) {
 	}
 	p.mu.Unlock()
 
+	// Interrupt takes precedence over any pending value or a normal Close.
+	// Without this pre-check the select below is chosen at random when both a
+	// value/closed channel and ctx.Done() are ready, which would let an
+	// abnormal termination be observed as a clean io.EOF — e.g. a producer
+	// that calls Interrupt() and then runs a deferred Close(). Checking
+	// ctx.Done() first makes "Interrupt returns context.Canceled immediately,
+	// skipping buffered values" a deterministic guarantee.
+	select {
+	case <-p.ctx.Done():
+		var zero T
+		err := p.ctx.Err()
+		p.mu.Lock()
+		p.lastErr = err
+		p.mu.Unlock()
+		return zero, err
+	default:
+	}
+
 	select {
 	case v, ok := <-p.ch:
 		if !ok {

voice/audio/stream_test.go

@@ -78,19 +78,14 @@ func TestPipe_InterruptSkipsBuffer(t *testing.T) {
 	}
 	pipe.Interrupt()
 
-	// Read must eventually return context.Canceled; it may return buffered values first
-	// (select is non-deterministic when both channel and ctx.Done() are ready).
-	var err error
-	for err == nil {
-		_, err = pipe.Read()
-	}
-	if err != context.Canceled {
+	// Interrupt is deterministic: the very first Read skips all buffered
+	// values and returns context.Canceled.
+	if _, err := pipe.Read(); err != context.Canceled {
 		t.Errorf("Read: got error %v, want context.Canceled", err)
 	}
 
 	// Subsequent reads also return the same error
-	_, err = pipe.Read()
-	if err != context.Canceled {
+	if _, err := pipe.Read(); err != context.Canceled {
 		t.Errorf("subsequent Read: got error %v, want context.Canceled", err)
 	}
 }
@@ -149,9 +144,10 @@ func TestPipe_InterruptThenClose(t *testing.T) {
 	pipe.Interrupt()
 	pipe.Close() // must not panic
 
-	// Read returns an error (either context.Canceled or io.EOF; select is non-deterministic when both are ready)
-	_, err := pipe.Read()
-	if err == nil {
-		t.Error("Read after Interrupt+Close: expected error, got nil")
+	// Interrupt takes precedence over Close: Read must deterministically
+	// report context.Canceled, never a clean io.EOF, so an abnormal
+	// termination is never observed as a normal end-of-stream.
+	if _, err := pipe.Read(); err != context.Canceled {
+		t.Errorf("Read after Interrupt+Close: got %v, want context.Canceled", err)
 	}
 }

voice/go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/pion/webrtc/v4 v4.2.9
 	github.com/rs/xid v1.6.0
 	go.opentelemetry.io/otel/metric v1.40.0
+	go.uber.org/goleak v1.3.0
 )
 
 require (

voice/provider/circuit_halfopen_test.go

@@ -0,0 +1,122 @@
+package provider
+
+import (
+	"testing"
+	"time"
+)
+
+// TestCircuit_HalfOpenConsecutiveFailuresStayBounded is a white-box regression
+// test for the half-open bounded-count fix (issue #4b).
+//
+// When a circuit's open window elapses, Allow transitions it to half-open and
+// decays consecutiveFailures to 0, and a probe failure re-opens the circuit
+// from that clean count. As a result the internal consecutiveFailures counter
+// must stay bounded no matter how many open -> probe-fail cycles occur.
+//
+// The existing black-box test (TestCircuit_HalfOpenProbeFailureReopensFromClean
+// Count) only checks a single re-open; it does not catch a counter that grows
+// without bound across many cycles. This test drives many cycles and asserts
+// the counter never exceeds the break threshold — before the fix it would grow
+// roughly linearly with the number of cycles.
+func TestCircuit_HalfOpenConsecutiveFailuresStayBounded(t *testing.T) {
+	c := NewCircuit(1)
+	policy := FallbackPolicy{CircuitBreakAfter: 2, CircuitOpen: 50 * time.Millisecond}
+	base := time.Now()
+
+	// Open the circuit initially: two failures reach the break threshold.
+	c.OnFailure(0, base, policy, true)
+	if !c.OnFailure(0, base, policy, true) {
+		t.Fatal("circuit should open after reaching CircuitBreakAfter")
+	}
+
+	maxSeen := 0
+	record := func() {
+		// Same-package white-box access to the internal counter.
+		c.mu.Lock()
+		got := c.states[0].consecutiveFailures
+		c.mu.Unlock()
+		if got > maxSeen {
+			maxSeen = got
+		}
+	}
+	record()
+
+	now := base
+	const cycles = 200
+	for cycle := 0; cycle < cycles; cycle++ {
+		// Let the open window elapse so Allow yields a half-open probe.
+		now = now.Add(60 * time.Millisecond)
+		if !c.Allow(0, now) {
+			t.Fatalf("cycle %d: expected a half-open probe to be allowed", cycle)
+		}
+		// The probe fails, which must re-open the circuit from a clean count.
+		if !c.OnFailure(0, now, policy, true) {
+			t.Fatalf("cycle %d: half-open probe failure should re-open the circuit", cycle)
+		}
+		record()
+	}
+
+	// Bounded property: the counter must not grow with the number of cycles.
+	// Before the fix it would be ~= 2 + cycles.
+	if maxSeen > policy.CircuitBreakAfter {
+		t.Fatalf("consecutiveFailures grew unbounded across %d cycles: max=%d, want <= %d",
+			cycles, maxSeen, policy.CircuitBreakAfter)
+	}
+}
+
+// TestCircuit_HalfOpenAllowsSingleProbe verifies that once the open window
+// elapses, Allow hands out exactly one half-open probe and holds every other
+// caller until the probe is resolved via OnSuccess/OnFailure. Before the fix,
+// the first probe zeroed openUntil and any concurrent/subsequent Allow saw a
+// zero window and returned true, letting a post-cooldown burst hit a possibly
+// still-unhealthy provider.
+func TestCircuit_HalfOpenAllowsSingleProbe(t *testing.T) {
+	policy := FallbackPolicy{CircuitBreakAfter: 1, CircuitOpen: 50 * time.Millisecond}
+	base := time.Now()
+
+	open := func() *Circuit {
+		c := NewCircuit(1)
+		if !c.OnFailure(0, base, policy, true) {
+			t.Fatal("circuit should open after reaching CircuitBreakAfter")
+		}
+		return c
+	}
+	afterCooldown := base.Add(60 * time.Millisecond)
+
+	t.Run("holds other callers until probe resolves", func(t *testing.T) {
+		c := open()
+		if !c.Allow(0, afterCooldown) {
+			t.Fatal("first caller after cooldown should get the half-open probe")
+		}
+		// Probe is in flight; all other callers must be held.
+		for i := 0; i < 5; i++ {
+			if c.Allow(0, afterCooldown) {
+				t.Fatalf("call %d: expected probe to be held while one is in flight", i)
+			}
+		}
+	})
+
+	t.Run("probe success reopens the gate", func(t *testing.T) {
+		c := open()
+		if !c.Allow(0, afterCooldown) {
+			t.Fatal("expected half-open probe")
+		}
+		c.OnSuccess(0)
+		if !c.Allow(0, afterCooldown) {
+			t.Fatal("after a successful probe the circuit should be closed and allow traffic")
+		}
+	})
+
+	t.Run("probe failure re-opens and blocks", func(t *testing.T) {
+		c := open()
+		if !c.Allow(0, afterCooldown) {
+			t.Fatal("expected half-open probe")
+		}
+		if !c.OnFailure(0, afterCooldown, policy, true) {
+			t.Fatal("failed probe should re-open the circuit")
+		}
+		if c.Allow(0, afterCooldown) {
+			t.Fatal("re-opened circuit should block until the new cooldown elapses")
+		}
+	})
+}

voice/provider/executor.go

@@ -14,9 +14,16 @@ import (
 var ErrSkipProvider = errors.New("provider: skip")
 
 // SleepWithContext blocks for duration d or until ctx is cancelled.
+// It reports true if the full duration elapsed and false if ctx was cancelled.
+// For d<=0 it still consults ctx so a cancelled context is honored immediately.
 func SleepWithContext(ctx context.Context, d time.Duration) bool {
 	if d <= 0 {
-		return true
+		select {
+		case <-ctx.Done():
+			return false
+		default:
+			return true
+		}
 	}
 	timer := time.NewTimer(d)
 	defer timer.Stop()
@@ -54,9 +61,21 @@ func RunWithFallback[T any](
 
 	var zero T
 	var lastErr error
-	now := time.Now()
 
 	for i := 0; i < n; i++ {
+		// Fail fast if the context is already done: return the context error
+		// directly so it is never misclassified as a provider failure that
+		// could trip the breaker.
+		if err := ctx.Err(); err != nil {
+			report.Error = err.Error()
+			emit()
+			return zero, err
+		}
+
+		// Recompute the clock per iteration: a slow multi-provider run can let
+		// a provider's open window elapse mid-loop, so a single pre-loop
+		// timestamp would keep skipping providers that are actually recovered.
+		now := time.Now()
 		name := nameFn(i)
 		if !circuit.Allow(i, now) {
 			report.Attempts = append(report.Attempts, Attempt{
@@ -69,6 +88,14 @@ func RunWithFallback[T any](
 
 		skipped := false
 		for attempt := 0; attempt < policy.MaxAttempts; attempt++ {
+			// Guard each attempt on the context so a cancelled/expired ctx
+			// stops retries deterministically even when RetryBackoff==0 (the
+			// backoff sleep is otherwise the only cancellation checkpoint).
+			if err := ctx.Err(); err != nil {
+				report.Error = err.Error()
+				emit()
+				return zero, err
+			}
 			result, err := execFn(ctx, i)
 			if err == nil {
 				circuit.OnSuccess(i)
@@ -91,7 +118,10 @@ func RunWithFallback[T any](
 			lastErr = err
 			retryable := policy.ShouldRetry(err)
 			fallbackable := policy.ShouldFallback(err)
-			circuitOpened := circuit.OnFailure(i, time.Now(), policy, retryable)
+			// Only provider-attributable faults count toward the breaker;
+			// client-fault errors (bad input, cancellation) must not open a
+			// healthy provider.
+			circuitOpened := circuit.OnFailure(i, time.Now(), policy, IsProviderFault(err))
 			report.Attempts = append(report.Attempts, Attempt{
 				Provider:      name,
 				Attempt:       attempt + 1,