fix: notify Go when SCStream stops unexpectedly and auto-restart speaker source (#5)

When macOS stops the ScreenCaptureKit stream (screen lock, display change,
system error), didStopWithError: was only logging — the Go channel stayed
open but received no data, causing silent hang and the menu-bar indicator
to disappear.

Changes:
- speaker_darwin.m: call tacitSpeakerStoppedCallback from didStopWithError:
  when the stream is not already in a deliberate-stop path (self.stopped)
- speaker_darwin.h: declare tacitSpeakerStoppedCallback
- speaker_darwin.go: replace bare chan[]int16 cgo handle with speakerChan
  (data channel + sync.Once + self-referencing handle) so both the
  unexpected-stop callback and the ctx.Done() goroutine can safely close
  the channel exactly once; add tacitSpeakerStoppedCallback export
- pipeline.go: split runSource into runSource (retry loop) + runSourceOnce
  so that an unexpected stream closure triggers automatic restart after 5s
  instead of silently stopping the pipeline

https://claude.ai/code/session_01LRhvpcLHSUZT1fz9XvbPny

Co-authored-by: Claude <noreply@anthropic.com>

Author: sangmin7648

pkg/capture/speaker_darwin.go

@@ -16,6 +16,7 @@ import (
 	"fmt"
 	"log"
 	"runtime/cgo"
+	"sync"
 	"unsafe"
 )
 
@@ -35,41 +36,56 @@ func NewSpeaker() (*Speaker, error) {
 	return &Speaker{}, nil
 }
 
+// speakerChan wraps the audio channel with a once-guarded close so both the
+// context-cancellation path and the unexpected-stop callback can safely signal
+// the pipeline without a double-close panic.
+type speakerChan struct {
+	data   chan []int16
+	handle cgo.Handle // points back to itself; used for self-deletion
+	once   sync.Once
+}
+
+func (sc *speakerChan) closeAndFree() {
+	sc.once.Do(func() {
+		close(sc.data)
+		sc.handle.Delete()
+	})
+}
+
 // Stream starts system-audio capture and returns a channel of int16 chunks.
-// The channel is closed when ctx is cancelled.
+// The channel is closed when ctx is cancelled or when SCStream stops unexpectedly.
 func (s *Speaker) Stream(ctx context.Context) (<-chan []int16, error) {
-	ch := make(chan []int16, 128)
-
-	// Register the channel in the cgo handle registry.
-	handle := cgo.NewHandle(ch)
+	sc := &speakerChan{data: make(chan []int16, 128)}
+	// Register sc in the cgo handle registry; the handle value is stored in sc
+	// itself so the ObjC callbacks can call closeAndFree via a single pointer.
+	sc.handle = cgo.NewHandle(sc)
 
 	var errCStr *C.char
-	cap := C.speaker_create(C.uintptr_t(handle), &errCStr)
+	cap := C.speaker_create(C.uintptr_t(sc.handle), &errCStr)
 	if errCStr != nil {
 		msg := C.GoString(errCStr)
 		C.free(unsafe.Pointer(errCStr))
-		handle.Delete()
-		close(ch)
+		sc.closeAndFree()
 		return nil, fmt.Errorf("speaker capture: %s", msg)
 	}
 	if cap == nil {
-		handle.Delete()
-		close(ch)
+		sc.closeAndFree()
 		return nil, fmt.Errorf("speaker capture: unknown error")
 	}
 
 	s.cap = cap
 
 	go func() {
 		<-ctx.Done()
+		// Set output.stopped = YES before stopping so didStopWithError won't
+		// fire the Go callback again after we deliberately stop the stream.
 		C.speaker_stop(cap)
 		s.cap = nil
-		handle.Delete()
-		close(ch)
+		sc.closeAndFree()
 	}()
 
 	log.Printf("System audio capture started (ScreenCaptureKit, 16kHz mono)")
-	return ch, nil
+	return sc.data, nil
 }
 
 // Close stops capture and releases resources if Stream was called.
@@ -86,7 +102,7 @@ func (s *Speaker) Close() {
 //export tacitSpeakerSamplesCallback
 func tacitSpeakerSamplesCallback(h C.uintptr_t, samples *C.int16_t, count C.int) {
 	handle := cgo.Handle(h)
-	ch, ok := handle.Value().(chan []int16)
+	sc, ok := handle.Value().(*speakerChan)
 	if !ok {
 		return
 	}
@@ -97,8 +113,24 @@ func tacitSpeakerSamplesCallback(h C.uintptr_t, samples *C.int16_t, count C.int)
 	copy(dst, buf)
 
 	select {
-	case ch <- dst:
+	case sc.data <- dst:
 	default:
 		// Drop frame if the pipeline is slow — prevents audio thread stall.
 	}
 }
+
+// tacitSpeakerStoppedCallback is called from Objective-C when SCStream stops
+// unexpectedly (not due to an explicit speaker_stop call).  Closing the channel
+// unblocks the pipeline's "for chunk := range stream" loop so it can detect
+// the outage and restart if desired.
+//
+//export tacitSpeakerStoppedCallback
+func tacitSpeakerStoppedCallback(h C.uintptr_t) {
+	handle := cgo.Handle(h)
+	sc, ok := handle.Value().(*speakerChan)
+	if !ok {
+		return
+	}
+	log.Printf("System audio capture: SCStream stopped unexpectedly, closing stream channel")
+	sc.closeAndFree()
+}

pkg/capture/speaker_darwin.h

@@ -9,6 +9,10 @@ typedef struct SpeakerCapture SpeakerCapture;
 // Declared here so the .m file can call it.
 extern void tacitSpeakerSamplesCallback(uintptr_t handle, int16_t* samples, int count);
 
+// tacitSpeakerStoppedCallback is called when SCStream stops unexpectedly.
+// Defined in speaker_darwin.go (//export).
+extern void tacitSpeakerStoppedCallback(uintptr_t handle);
+
 // speaker_create starts ScreenCaptureKit system audio capture (macOS 13+).
 // On success returns a non-NULL handle and sets *errMsg to NULL.
 // On failure returns NULL and sets *errMsg to a malloc'd error string (caller must free).

pkg/capture/speaker_darwin.m

@@ -84,8 +84,14 @@ - (void)stream:(SCStream *)stream
 }
 
 - (void)stream:(SCStream *)stream didStopWithError:(NSError *)error API_AVAILABLE(macos(12.3)) {
-    if (error && !self.stopped) {
-        NSLog(@"[tacit] speaker stream stopped with error: %@", error.localizedDescription);
+    if (!self.stopped) {
+        if (error) {
+            NSLog(@"[tacit] speaker stream stopped with error: %@", error.localizedDescription);
+        } else {
+            NSLog(@"[tacit] speaker stream stopped unexpectedly (no error)");
+        }
+        // Notify Go so it can close the channel and let the pipeline restart.
+        tacitSpeakerStoppedCallback(self.goHandle);
     }
 }
 

pkg/pipeline/pipeline.go

@@ -117,9 +117,32 @@ func (p *Pipeline) Run(ctx context.Context, sources []capture.AudioSource, label
 }
 
 // runSource runs a single audio source through VAD→STT and enqueues results
-// onto classifyCh.  It returns when ctx is cancelled or the source stream
-// closes.
+// onto classifyCh.  It retries automatically when the source stream closes
+// unexpectedly (e.g. SCStream stopped by macOS), and returns only when ctx is
+// cancelled or a fatal initialisation error occurs.
 func (p *Pipeline) runSource(ctx context.Context, src capture.AudioSource, label string, classifyCh chan<- classifyItem) error {
+	const retryDelay = 5 * time.Second
+	for {
+		err := p.runSourceOnce(ctx, src, label, classifyCh)
+		if err != nil {
+			return err
+		}
+		if ctx.Err() != nil {
+			return nil
+		}
+		// Stream closed unexpectedly — wait briefly then restart.
+		log.Printf("[%s] stream closed unexpectedly, restarting in %v", label, retryDelay)
+		select {
+		case <-time.After(retryDelay):
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+// runSourceOnce runs one capture session for a source.  It returns when ctx is
+// cancelled or the source's stream channel is closed (normal or unexpected).
+func (p *Pipeline) runSourceOnce(ctx context.Context, src capture.AudioSource, label string, classifyCh chan<- classifyItem) error {
 	// Init per-source VAD (256 samples = 16 ms at 16 kHz).
 	const hopSize = 256
 	v, err := vad.New(hopSize, float32(p.cfg.SpeechThreshold))
@@ -130,6 +153,9 @@ func (p *Pipeline) runSource(ctx context.Context, src capture.AudioSource, label
 
 	stream, err := src.Stream(ctx)
 	if err != nil {
+		if ctx.Err() != nil {
+			return nil // ctx cancelled during restart, not a real error
+		}
 		return fmt.Errorf("start stream: %w", err)
 	}