Maven/internal/stt/pair.go

package stt

import (
	"context"
	"errors"
	"log"
	"net/http"
	"sync"
	"sync/atomic"
	"time"

	"github.com/kami/maven/internal/audio"
)

// Pair — a preferred transcriber on the workstation, with mavsttd as the floor.
//
// Same arrangement as llm.Pair and for the same reason. The microphone is at
// workpc, the card there has 16GB, and CrisperWhisper 2.0 turbo scores 10.4%
// WER in Russian against 27.5% for the ggml-small.bin homesrv loads
// (docs/evals/2026-08-09-crisperwhisper2-russian-wer.md). The workstation is
// never assumed up: it sleeps, and the card is often held by a training run.
//
// Speech-to-text has only the silent half of the degradation rule. A worse
// transcript is still a turn, and there is nothing to name a gap about, so
// Transcribe always falls back. That is the whole difference from llm.Pair,
// which also carries CompleteRemote for callers that must refuse instead.
type Pair struct {
	remote Transcriber
	floor  Transcriber

	// up — the cached admission answer, written only by the prober and read by
	// every turn. A voice turn must never wait on a machine that may be asleep.
	up atomic.Bool

	health   string
	interval time.Duration
	http     *http.Client
	stop     chan struct{}
	stopOnce sync.Once
}

const (
	probeTimeout         = 2 * time.Second
	defaultProbeInterval = 15 * time.Second
)

// ErrNoFloor — a Pair was built with no local transcriber to fall back to. A
// configuration mistake: the floor is what makes the remote optional.
var ErrNoFloor = errors.New("stt: no floor transcriber")

// NewPair builds the two-transcriber arrangement. remote may be nil, which is
// the unconfigured deploy: every turn goes to the floor and nothing probes.
func NewPair(remote, floor Transcriber, health string, interval time.Duration) *Pair {
	if interval <= 0 {
		// The config normalises this, so a zero here is a caller that built the
		// Pair directly. Panicking in a ticker is the wrong way to say so.
		interval = defaultProbeInterval
	}
	return &Pair{
		remote:   remote,
		floor:    floor,
		health:   health,
		interval: interval,
		http:     &http.Client{Timeout: probeTimeout},
		stop:     make(chan struct{}),
	}
}

// Start begins probing. The first probe runs before the first tick, so a
// workstation that is already up serves the first utterance rather than the
// second. Safe with a nil remote.
func (p *Pair) Start(ctx context.Context) {
	if p.remote == nil || p.health == "" {
		return
	}
	go func() {
		p.probe(ctx)
		t := time.NewTicker(p.interval)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-p.stop:
				return
			case <-t.C:
				p.probe(ctx)
			}
		}
	}()
}

// Stop ends the prober. Idempotent and safe from two goroutines.
func (p *Pair) Stop() {
	p.stopOnce.Do(func() { close(p.stop) })
}

// Available reports whether the workstation will transcribe right now.
func (p *Pair) Available() bool {
	return p.remote != nil && p.up.Load()
}

func (p *Pair) probe(ctx context.Context) {
	ctx, cancel := context.WithTimeout(ctx, probeTimeout)
	defer cancel()
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, p.health, nil)
	if err != nil {
		p.set(false)
		return
	}
	resp, err := p.http.Do(req)
	if err != nil {
		p.set(false)
		return
	}
	defer resp.Body.Close()
	p.set(resp.StatusCode == http.StatusOK)
}

// set records the admission answer and logs only transitions. A machine that
// sleeps nightly would otherwise write one line per interval forever.
func (p *Pair) set(up bool) {
	if p.up.Swap(up) == up {
		return
	}
	if up {
		log.Printf("stt: workstation transcriber available at %s", p.health)
	} else {
		log.Print("stt: workstation transcriber unavailable, falling back to mavsttd")
	}
}

// Transcribe sends the audio to the workstation when it will take work, and to
// mavsttd otherwise. A remote that fails mid-request falls back too, because
// the admission answer is a cache and can be one interval out of date.
//
// Killing the remote mid-session must not drop the turn. That is the whole
// point of the floor, and it is what TestPairFallsBackWhenRemoteFails pins.
func (p *Pair) Transcribe(ctx context.Context, a audio.Audio) (string, float64, error) {
	if p.floor == nil {
		return "", 0, ErrNoFloor
	}
	if p.Available() {
		text, conf, err := p.remote.Transcribe(ctx, a)
		if err == nil {
			err = validateTranscript(text, conf)
		}
		if err == nil {
			log.Print("stt: transcribed on the workstation")
			return text, conf, nil
		}
		// The cached answer was wrong. Correct it now rather than sending the
		// next utterance into the same hole, then fall back.
		p.set(false)
		log.Printf("stt: workstation failed mid-request, falling back: %v", err)
	}
	return p.floor.Transcribe(ctx, a)
}

var _ Transcriber = (*Pair)(nil)