The workstation transcribes, homesrv is the floor (V-486)
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. The workstation is never assumed up: it sleeps, and the card is often held. Admission is a cached atomic written only by the prober, so no voice turn ever waits on a machine that may be asleep. Speech-to-text has only the silent half of the degradation rule. A worse transcript is still a turn, so there is nothing to name a gap about and Transcribe always falls back. That is the whole difference from llm.Pair, which also carries CompleteRemote for callers that must refuse instead. A remote that dies mid-request corrects the cache and falls back in the same turn, which is what TestPairFallsBackWhenRemoteFails pins. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013ptwopxyo3Z2kwFckHkLvN
This commit is contained in:
@@ -0,0 +1,157 @@
|
||||
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 {
|
||||
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)
|
||||
@@ -0,0 +1,143 @@
|
||||
package stt
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kami/maven/internal/audio"
|
||||
)
|
||||
|
||||
// scripted — a Transcriber that answers with a fixed text, or fails.
|
||||
type scripted struct {
|
||||
text string
|
||||
err error
|
||||
calls atomic.Int32
|
||||
}
|
||||
|
||||
func (s *scripted) Transcribe(_ context.Context, _ audio.Audio) (string, float64, error) {
|
||||
s.calls.Add(1)
|
||||
if s.err != nil {
|
||||
return "", 0, s.err
|
||||
}
|
||||
return s.text, 0.9, nil
|
||||
}
|
||||
|
||||
func sample() audio.Audio {
|
||||
return audio.Audio{Format: audio.PCM16kMono, Bytes: make([]byte, 3200)}
|
||||
}
|
||||
|
||||
// up builds a Pair whose admission answer is already true, without probing.
|
||||
func up(remote, floor Transcriber) *Pair {
|
||||
p := NewPair(remote, floor, "", time.Minute)
|
||||
p.up.Store(true)
|
||||
return p
|
||||
}
|
||||
|
||||
func TestPairPrefersTheWorkstation(t *testing.T) {
|
||||
t.Parallel()
|
||||
remote := &scripted{text: "с рабочей станции"}
|
||||
floor := &scripted{text: "с homesrv"}
|
||||
text, _, err := up(remote, floor).Transcribe(context.Background(), sample())
|
||||
if err != nil {
|
||||
t.Fatalf("Transcribe: %v", err)
|
||||
}
|
||||
if text != "с рабочей станции" {
|
||||
t.Fatalf("want the remote transcript, got %q", text)
|
||||
}
|
||||
if floor.calls.Load() != 0 {
|
||||
t.Fatalf("floor was called %d times, want 0", floor.calls.Load())
|
||||
}
|
||||
}
|
||||
|
||||
// The turn is what matters. A remote that dies mid-session must cost a worse
|
||||
// transcript and nothing else. This is the V-486 bar.
|
||||
func TestPairFallsBackWhenRemoteFails(t *testing.T) {
|
||||
t.Parallel()
|
||||
remote := &scripted{err: errors.New("connection refused")}
|
||||
floor := &scripted{text: "с homesrv"}
|
||||
p := up(remote, floor)
|
||||
|
||||
text, conf, err := p.Transcribe(context.Background(), sample())
|
||||
if err != nil {
|
||||
t.Fatalf("a failed remote must not fail the turn: %v", err)
|
||||
}
|
||||
if text != "с homesrv" {
|
||||
t.Fatalf("want the floor transcript, got %q", text)
|
||||
}
|
||||
if conf != 0.9 {
|
||||
t.Fatalf("want the floor confidence, got %v", conf)
|
||||
}
|
||||
if p.Available() {
|
||||
t.Fatal("a failed request must correct the cached admission answer")
|
||||
}
|
||||
|
||||
// The next utterance goes straight to the floor rather than into the
|
||||
// same hole.
|
||||
if _, _, err := p.Transcribe(context.Background(), sample()); err != nil {
|
||||
t.Fatalf("second turn: %v", err)
|
||||
}
|
||||
if remote.calls.Load() != 1 {
|
||||
t.Fatalf("remote called %d times, want 1", remote.calls.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestPairWithNoRemoteIsTheFloor(t *testing.T) {
|
||||
t.Parallel()
|
||||
floor := &scripted{text: "с homesrv"}
|
||||
p := NewPair(nil, floor, "", time.Minute)
|
||||
p.Start(context.Background()) // no health url, so this is a no-op
|
||||
if p.Available() {
|
||||
t.Fatal("an unconfigured remote is never available")
|
||||
}
|
||||
text, _, err := p.Transcribe(context.Background(), sample())
|
||||
if err != nil {
|
||||
t.Fatalf("Transcribe: %v", err)
|
||||
}
|
||||
if text != "с homesrv" {
|
||||
t.Fatalf("want the floor transcript, got %q", text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPairWithNoFloorRefuses(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, _, err := NewPair(nil, nil, "", time.Minute).Transcribe(context.Background(), sample())
|
||||
if !errors.Is(err, ErrNoFloor) {
|
||||
t.Fatalf("want ErrNoFloor, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPairProbeReadsHealth(t *testing.T) {
|
||||
t.Parallel()
|
||||
var ok atomic.Bool
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
if !ok.Load() {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
p := NewPair(&scripted{text: "remote"}, &scripted{text: "floor"}, srv.URL, time.Minute)
|
||||
p.probe(context.Background())
|
||||
if p.Available() {
|
||||
t.Fatal("a 503 means the card is busy, so the workstation is not available")
|
||||
}
|
||||
ok.Store(true)
|
||||
p.probe(context.Background())
|
||||
if !p.Available() {
|
||||
t.Fatal("a 200 means the workstation will take work")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPairStopIsIdempotent(t *testing.T) {
|
||||
t.Parallel()
|
||||
p := NewPair(nil, &scripted{}, "", time.Minute)
|
||||
p.Stop()
|
||||
p.Stop()
|
||||
}
|
||||
Reference in New Issue
Block a user