fix: drain Brabble subprocess output before waiting

CHANGELOG.md

@@ -7,3 +7,4 @@
 - Treat leading dashes in spoken text as words instead of espeak options. Thanks @SebTardif! (#11)
 - Cancel bridge dialing, pairing, and hello waits on SIGINT/SIGTERM. Thanks @SebTardif! (#6)
 - Interrupt bridge reconnect backoff promptly on SIGINT/SIGTERM. Thanks @SebTardif! (#5)
+- Drain Brabble output before reaping the process and keep stderr flowing with disabled logging or oversized diagnostics, preventing lost transcripts and stalled recognition.

README.md

@@ -103,3 +103,4 @@ Override to `_clawdbot-bridge._tcp` if you intentionally want it to show up as a
 - Node state (`nodeId` + token) lives in `~/.clawdbot/clawgo.json`.
 - Caps default to `voiceWake`; override via `-caps` if you expose more commands.
 - Set `bridge.bind: "tailnet"` on the gateway to restrict the bridge to Tailscale.
+- Brabble output is drained before its transcript stream closes; stderr diagnostics cannot block recognition when logging is disabled or a diagnostic line is oversized. Nonzero process exits are logged.

modules/stt/brabble.go

@@ -35,31 +35,49 @@ func (e *BrabbleEngine) Transcribe(ctx context.Context, _ <-chan audio.Frame, _
 	if cmdPath == "" {
 		cmdPath = "brabble"
 	}
+	ctx, cancel := context.WithCancel(ctx)
 	cmd := exec.CommandContext(ctx, cmdPath, e.cfg.Args...)
 	stdout, err := cmd.StdoutPipe()
 	if err != nil {
+		cancel()
 		return nil, err
 	}
 	stderr, err := cmd.StderrPipe()
 	if err != nil {
+		cancel()
 		return nil, err
 	}
 	if err := cmd.Start(); err != nil {
+		cancel()
 		return nil, err
 	}
+	stopClosing := context.AfterFunc(ctx, func() {
+		_ = stdout.Close()
+		_ = stderr.Close()
+	})
 	out := make(chan Transcript, 32)
+	stderrDone := make(chan struct{})
 	go func() {
-		defer close(out)
-		e.readLines(ctx, stdout, out)
+		defer close(stderrDone)
+		e.logLines(ctx, stderr)
 	}()
-	go e.logLines(ctx, stderr)
 	go func() {
-		_ = cmd.Wait()
+		defer close(out)
+		defer cancel()
+		defer stopClosing()
+		if err := e.readLines(ctx, stdout, out); err != nil {
+			cancel()
+		}
+		<-stderrDone
+		// Wait closes the pipes, so both readers must finish first.
+		if err := cmd.Wait(); err != nil && ctx.Err() == nil && e.logf != nil {
+			e.logf("brabble exited: %v", err)
+		}
 	}()
 	return out, nil
 }
 
-func (e *BrabbleEngine) readLines(ctx context.Context, r io.Reader, out chan<- Transcript) {
+func (e *BrabbleEngine) readLines(ctx context.Context, r io.Reader, out chan<- Transcript) error {
 	scanner := bufio.NewScanner(r)
 	scanner.Buffer(make([]byte, 0, 64*1024), 8*1024*1024)
 	for scanner.Scan() {
@@ -74,19 +92,22 @@ func (e *BrabbleEngine) readLines(ctx context.Context, r io.Reader, out chan<- T
 		select {
 		case out <- tr:
 		case <-ctx.Done():
-			return
+			return ctx.Err()
 		}
 	}
-	if err := scanner.Err(); err != nil && !errors.Is(err, io.EOF) && e.logf != nil {
+	if err := scanner.Err(); err != nil && !errors.Is(err, io.EOF) && ctx.Err() == nil && e.logf != nil {
 		e.logf("brabble read error: %v", err)
 	}
+	return scanner.Err()
 }
 
 func (e *BrabbleEngine) logLines(ctx context.Context, r io.Reader) {
 	if e.logf == nil {
+		_, _ = io.Copy(io.Discard, r)
 		return
 	}
 	scanner := bufio.NewScanner(r)
+	scanner.Buffer(make([]byte, 0, 64*1024), 8*1024*1024)
 	for scanner.Scan() {
 		select {
 		case <-ctx.Done():
@@ -99,6 +120,11 @@ func (e *BrabbleEngine) logLines(ctx context.Context, r io.Reader) {
 		}
 		e.logf("brabble: %s", line)
 	}
+	if err := scanner.Err(); err != nil && ctx.Err() == nil {
+		e.logf("brabble stderr read error: %v", err)
+		// Keep draining even when a diagnostic exceeds the line limit.
+		_, _ = io.Copy(io.Discard, r)
+	}
 }
 
 type brabbleEvent struct {

modules/stt/brabble_process_test.go

@@ -0,0 +1,113 @@
+package stt
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestBrabbleProcess(t *testing.T) {
+	if len(os.Args) < 2 {
+		return
+	}
+	switch os.Args[len(os.Args)-1] {
+	case "brabble-output":
+		for i := 0; i < 500; i++ {
+			fmt.Printf("%03d %s\n", i, strings.Repeat("x", 100))
+		}
+		os.Exit(0)
+	case "brabble-stderr":
+		fmt.Fprintln(os.Stderr, strings.Repeat("x", 9*1024*1024))
+		fmt.Println("complete")
+		os.Exit(0)
+	}
+}
+
+func TestBrabbleCancellationUnblocksOutput(t *testing.T) {
+	executable, err := os.Executable()
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	engine := NewBrabbleEngine(BrabbleConfig{Command: executable, Args: []string{"-test.run=^TestBrabbleProcess$", "--", "brabble-output"}}, nil)
+	out, err := engine.Transcribe(ctx, nil, Options{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-out:
+	case <-time.After(5 * time.Second):
+		t.Fatal("recognizer did not start")
+	}
+	cancel()
+	done := make(chan struct{})
+	go func() {
+		for range out {
+		}
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("canceled recognizer did not close its output")
+	}
+}
+
+func TestBrabbleDrainsOutputBeforeWaiting(t *testing.T) {
+	executable, err := os.Executable()
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	engine := NewBrabbleEngine(BrabbleConfig{Command: executable, Args: []string{"-test.run=^TestBrabbleProcess$", "--", "brabble-output"}}, nil)
+	out, err := engine.Transcribe(ctx, nil, Options{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	count := 0
+	for tr := range out {
+		want := fmt.Sprintf("%03d %s", count, strings.Repeat("x", 100))
+		if tr.Text != want {
+			t.Fatalf("transcript %d = %q, want %q", count, tr.Text, want)
+		}
+		count++
+		time.Sleep(time.Millisecond)
+	}
+	if count != 500 {
+		t.Fatalf("delivered %d of 500 transcripts", count)
+	}
+}
+
+func TestBrabbleDrainsStderr(t *testing.T) {
+	executable, err := os.Executable()
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, logging := range []bool{false, true} {
+		t.Run(fmt.Sprintf("logging=%v", logging), func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer cancel()
+			var logf func(string, ...any)
+			if logging {
+				logf = func(string, ...any) {}
+			}
+			engine := NewBrabbleEngine(BrabbleConfig{Command: executable, Args: []string{"-test.run=^TestBrabbleProcess$", "--", "brabble-stderr"}}, logf)
+			out, err := engine.Transcribe(ctx, nil, Options{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			var transcripts []string
+			for tr := range out {
+				transcripts = append(transcripts, tr.Text)
+			}
+			if len(transcripts) != 1 || transcripts[0] != "complete" {
+				t.Fatalf("stderr stalled stdout: %v (context: %v)", transcripts, ctx.Err())
+			}
+		})
+	}
+}