diff --git a/internal/stt/pair.go b/internal/stt/pair.go new file mode 100644 index 0000000..2001ed6 --- /dev/null +++ b/internal/stt/pair.go @@ -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) diff --git a/internal/stt/pair_test.go b/internal/stt/pair_test.go new file mode 100644 index 0000000..b14b241 --- /dev/null +++ b/internal/stt/pair_test.go @@ -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() +}