feat(capture): dynamic sink-level audio capture

internal/audio/pcm/pcm.go

@@ -45,3 +45,32 @@ func TrimTrailingFrames(pcm []byte, frames, frameBytes int) []byte {
 	}
 	return pcm
 }
+
+// Mix combines s16le mono PCM frames via saturating addition. Silent or idle
+// inputs (all-zero) don't attenuate active ones the way averaging would.
+// Frames are expected to be FrameBytes long; shorter frames contribute only
+// their available samples. Mixing zero frames returns FrameBytes of silence.
+func Mix(frames ...[]byte) []byte {
+	out := make([]byte, FrameBytes)
+	for _, f := range frames {
+		n := min(len(f), FrameBytes) / 2
+		for i := range n {
+			a := int16(binary.LittleEndian.Uint16(out[i*2 : i*2+2]))
+			b := int16(binary.LittleEndian.Uint16(f[i*2 : i*2+2]))
+			binary.LittleEndian.PutUint16(out[i*2:i*2+2], uint16(saturatingAdd(a, b)))
+		}
+	}
+	return out
+}
+
+func saturatingAdd(a, b int16) int16 {
+	sum := int32(a) + int32(b)
+	switch {
+	case sum > math.MaxInt16:
+		return math.MaxInt16
+	case sum < math.MinInt16:
+		return math.MinInt16
+	default:
+		return int16(sum)
+	}
+}

internal/audio/pcm/pcm_test.go

@@ -1,8 +1,10 @@
 package pcm
 
 import (
+	"bytes"
 	"encoding/binary"
 	"math"
+	"slices"
 	"testing"
 )
 
@@ -71,6 +73,94 @@ func TestFrameCount(t *testing.T) {
 	}
 }
 
+func sampleFrame(t *testing.T, samples ...int16) []byte {
+	t.Helper()
+	buf := make([]byte, FrameBytes)
+	for i, s := range samples {
+		binary.LittleEndian.PutUint16(buf[i*2:], uint16(s))
+	}
+	return buf
+}
+
+func readSamples(t *testing.T, data []byte, n int) []int16 {
+	t.Helper()
+	out := make([]int16, n)
+	for i := range n {
+		out[i] = int16(binary.LittleEndian.Uint16(data[i*2:]))
+	}
+	return out
+}
+
+func TestMix_NoInputs(t *testing.T) {
+	got := Mix()
+	if len(got) != FrameBytes {
+		t.Fatalf("len = %d, want %d", len(got), FrameBytes)
+	}
+	for _, b := range got {
+		if b != 0 {
+			t.Fatalf("expected silence, got non-zero byte")
+		}
+	}
+}
+
+func TestMix_SingleInput(t *testing.T) {
+	f := sampleFrame(t, 100, -200, 300)
+	got := Mix(f)
+	if !bytes.Equal(got, f) {
+		t.Errorf("single input should be unchanged")
+	}
+}
+
+func TestMix_TwoInputsAdd(t *testing.T) {
+	a := sampleFrame(t, 100, -200, 300)
+	b := sampleFrame(t, 50, -50, -100)
+	got := Mix(a, b)
+	want := []int16{150, -250, 200}
+	if s := readSamples(t, got, len(want)); !slices.Equal(s, want) {
+		t.Errorf("Mix() = %v, want %v", s, want)
+	}
+}
+
+func TestMix_SilentDoesNotAttenuate(t *testing.T) {
+	active := sampleFrame(t, 1000, -1000, 500)
+	silent := make([]byte, FrameBytes)
+	got := Mix(active, silent)
+	if !bytes.Equal(got, active) {
+		t.Errorf("mixing with silence should leave active frame unchanged")
+	}
+}
+
+func TestMix_PositiveOverflowClamps(t *testing.T) {
+	a := sampleFrame(t, math.MaxInt16)
+	b := sampleFrame(t, math.MaxInt16)
+	got := Mix(a, b)
+	want := []int16{math.MaxInt16}
+	if s := readSamples(t, got, 1); !slices.Equal(s, want) {
+		t.Errorf("Mix() = %v, want %v", s, want)
+	}
+}
+
+func TestMix_NegativeOverflowClamps(t *testing.T) {
+	a := sampleFrame(t, math.MinInt16)
+	b := sampleFrame(t, math.MinInt16)
+	got := Mix(a, b)
+	want := []int16{math.MinInt16}
+	if s := readSamples(t, got, 1); !slices.Equal(s, want) {
+		t.Errorf("Mix() = %v, want %v", s, want)
+	}
+}
+
+func TestMix_InputBuffersUnchanged(t *testing.T) {
+	a := sampleFrame(t, 100, -200)
+	b := sampleFrame(t, 50, -50)
+	aCopy := append([]byte(nil), a...)
+	bCopy := append([]byte(nil), b...)
+	Mix(a, b)
+	if !bytes.Equal(a, aCopy) || !bytes.Equal(b, bCopy) {
+		t.Errorf("Mix() must not mutate its input buffers")
+	}
+}
+
 func TestTrimTrailingFrames(t *testing.T) {
 	pcm := make([]byte, FrameBytes*3)
 	tests := []struct {

internal/capture/capture.go

@@ -4,12 +4,19 @@ import (
 	"context"
 
 	"github.com/odsod/recorder/internal/audio/frame"
+	"github.com/odsod/recorder/internal/protocol/parec"
 )
 
 // Source abstracts dual-channel audio capture (system + microphone).
 type Source interface {
 	Start(ctx context.Context) (<-chan frame.Dual, error)
 	Stop() error
-	MonitorSource() string
-	MicSource() string
+}
+
+// sinkClient is the subset of *parec.Client that capture depends on.
+// Narrowed for testability; *parec.Client satisfies it structurally.
+type sinkClient interface {
+	ListSinks(ctx context.Context, req parec.ListSinksRequest) (parec.ListSinksResponse, error)
+	GetDefaultSource(ctx context.Context, req parec.GetDefaultSourceRequest) (parec.GetDefaultSourceResponse, error)
+	StartCapture(ctx context.Context, req parec.StartCaptureRequest) (*parec.CaptureStream, error)
 }

internal/capture/parec.go

@@ -2,119 +2,116 @@ package capture
 
 import (
 	"context"
-	"errors"
-	"fmt"
-	"io"
 	"sync"
+	"time"
 
 	"github.com/odsod/recorder/internal/audio/frame"
 	"github.com/odsod/recorder/internal/audio/pcm"
 	"github.com/odsod/recorder/internal/protocol/parec"
 )
 
-// Parec implements Source using PulseAudio's parec command.
+// tickInterval is the mix loop's emission period, matching pcm.FrameBytes'
+// one second of audio. A package-level var so tests can shrink it.
+var tickInterval = time.Second
+
+// Parec implements Source by dynamically monitoring every PulseAudio sink
+// plus the default microphone, mixing simultaneous sink audio together.
 type Parec struct {
-	client  *parec.Client
-	monitor string
-	mic     string
-	stop    func()
-}
+	client sinkClient
 
-// NewParec creates a Parec source using the given parec protocol client.
-func NewParec(client *parec.Client) *Parec {
-	return &Parec{client: client}
-}
+	mu      sync.Mutex
+	sinks   map[string]*reader // keyed by sink name
+	mic     *reader
+	micName string
+
+	sinkBackoff backoff
+	sinkLastTry time.Time
+	micBackoff  backoff
+	micLastTry  time.Time
 
-// MonitorSource returns the system audio monitor source name.
-func (c *Parec) MonitorSource() string {
-	return c.monitor
+	cancel   context.CancelFunc
+	wg       sync.WaitGroup
+	stopOnce sync.Once
 }
 
-// MicSource returns the microphone source name.
-func (c *Parec) MicSource() string {
-	return c.mic
+// NewParec creates a Parec source using the given parec protocol client.
+func NewParec(client *parec.Client) *Parec {
+	return &Parec{client: client, sinks: make(map[string]*reader)}
 }
 
-// Start begins capturing system and microphone audio, returning a channel of frames.
+// Start begins dynamic sink and microphone capture, returning a channel of
+// mixed frames. Discovery and stream startup happen in the background;
+// outages (no sinks, unreachable PulseAudio, a dead parec process) surface
+// as silent frames rather than failing Start or closing the channel — the
+// channel only closes once ctx is done.
 func (c *Parec) Start(ctx context.Context) (<-chan frame.Dual, error) {
-	sinkResp, err := c.client.GetDefaultSink(ctx, parec.GetDefaultSinkRequest{})
-	if err != nil {
-		return nil, err
-	}
-	sourceResp, err := c.client.GetDefaultSource(ctx, parec.GetDefaultSourceRequest{})
-	if err != nil {
-		return nil, err
-	}
-	c.monitor = sinkResp.MonitorSource
-	c.mic = sourceResp.Source
-
-	sysStream, err := c.client.StartCapture(ctx, parec.StartCaptureRequest{
-		Device: c.monitor, SampleRate: pcm.SampleRate,
-	})
-	if err != nil {
-		return nil, fmt.Errorf("start sys parec: %w", err)
-	}
-	micStream, err := c.client.StartCapture(ctx, parec.StartCaptureRequest{
-		Device: c.mic, SampleRate: pcm.SampleRate,
-	})
-	if err != nil {
-		_ = sysStream.Close()
-		return nil, fmt.Errorf("start mic parec: %w", err)
-	}
+	runCtx, cancel := context.WithCancel(ctx)
+	c.cancel = cancel
 
 	frames := make(chan frame.Dual, 2)
-	done := make(chan struct{})
-	var once sync.Once
-	c.stop = func() {
-		once.Do(func() {
-			close(done)
-			_ = sysStream.Close()
-			_ = micStream.Close()
-		})
-	}
 
-	go func() {
-		defer close(frames)
-		defer c.stop()
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-done:
-				return
-			default:
-			}
+	c.wg.Go(func() { c.reconcileLoop(runCtx) })
+	c.wg.Go(func() { c.mixLoop(runCtx, frames) })
 
-			sysData, err := frame.Read(sysStream, pcm.FrameBytes)
-			if err != nil {
-				if !errors.Is(err, io.EOF) {
-					return
-				}
-				return
-			}
-			micData, err := frame.Read(micStream, pcm.FrameBytes)
-			if err != nil {
-				micData = frame.Silent(pcm.FrameBytes)
-			}
+	return frames, nil
+}
 
-			f := frame.Dual{Sys: sysData, Mic: micData}
+func (c *Parec) mixLoop(ctx context.Context, out chan<- frame.Dual) {
+	defer close(out)
+	ticker := time.NewTicker(tickInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			f := c.mixTick()
 			select {
-			case frames <- f:
+			case out <- f:
 			case <-ctx.Done():
 				return
-			case <-done:
-				return
 			}
 		}
-	}()
+	}
+}
 
-	return frames, nil
+func (c *Parec) mixTick() frame.Dual {
+	c.mu.Lock()
+	sinkFrames := make([][]byte, 0, len(c.sinks))
+	for _, r := range c.sinks {
+		if data, ok := r.take(); ok {
+			sinkFrames = append(sinkFrames, data)
+		}
+	}
+	micReader := c.mic
+	c.mu.Unlock()
+
+	mic := frame.Silent(pcm.FrameBytes)
+	if micReader != nil {
+		if data, ok := micReader.take(); ok {
+			mic = data
+		}
+	}
+
+	return frame.Dual{Sys: pcm.Mix(sinkFrames...), Mic: mic}
 }
 
-// Stop terminates the capture processes.
+// Stop terminates all capture streams. Idempotent.
 func (c *Parec) Stop() error {
-	if c.stop != nil {
-		c.stop()
-	}
+	c.stopOnce.Do(func() {
+		if c.cancel != nil {
+			c.cancel()
+		}
+		c.wg.Wait()
+
+		c.mu.Lock()
+		for _, r := range c.sinks {
+			_ = r.stop()
+		}
+		if c.mic != nil {
+			_ = c.mic.stop()
+		}
+		c.mu.Unlock()
+	})
 	return nil
 }