diff --git a/internal/capture/capture.go b/internal/capture/capture.go index c9cb852..9f25979 100644 --- a/internal/capture/capture.go +++ b/internal/capture/capture.go @@ -16,8 +16,13 @@ // plan document and is refused: it requires listening in order to notice the // keyword, which is the exact behaviour this capability must not have. // - A session that is not stopped stops itself. MaxDuration is a hard cap -// checked on every Append, not a suggestion; a forgotten recording is a -// recording that ends, not one that runs until the disk is full. +// checked on every Append AND against the wall clock in Start and Status, +// so a client that simply stops sending frames — a browser tab closed, wifi +// gone — does not leave the one session slot occupied until mavend +// restarts. +// - A session belongs to whoever started it. Start returns a token and Append +// and Stop require it, so a second surface at the same authority rung +// cannot feed or harvest a recording it did not begin. // - Audio is stored under internal/media, which means retention prunes it and // it never leaves the box. Both the audio blob and the transcript stay // local; only the summary is written where he will read it. @@ -37,16 +42,20 @@ // // There is exactly one STT in Maven and this package does not add a second: it // takes an stt.Transcriber, which in deploy is the whisper.cpp worker behind -// cmd/mavsttd. Long audio is transcribed in windows too (see chunkAudio), for +// cmd/mavsttd. Long audio is transcribed in windows too (see transcribeFile), for // the same reason whisper itself works in 30s windows — handing a worker an hour // of PCM in one call is a request that either times out or blocks everything -// else for minutes. +// else for minutes. The windows are read back off the stored WAV one at a time, +// so the meeting is never in memory whole. package capture import ( "context" + "crypto/rand" + "encoding/hex" "errors" "fmt" + "os" "strings" "sync" "time" @@ -58,12 +67,17 @@ import ( // DefaultMaxDuration — how long one capture may run before it stops itself. // Two hours covers a long meeting and bounds the damage of a forgotten session: -// at 16 kHz mono that is about 230 MB of PCM, which is over media's default -// per-blob cap, so a session at the limit is stored truncated rather than -// refused. That trade is deliberate — a partial recording of a meeting he asked -// for beats an error after two hours. +// at 16 kHz mono that is about 230 MB of WAV, which is under media's +// DefaultMaxAudioBytes of 512 MiB. The two constants used to disagree — a +// 64 MiB blob cap is 35 minutes of audio against a 120 minute session cap — so +// the meeting that hit the limit was the one that failed to store. const DefaultMaxDuration = 2 * time.Hour +// StaleGrace — how long past MaxDuration a session may sit before Start and +// Status reap it. A frame in flight when the cap fires should not race the +// reaper, and a minute of slack costs nothing against a two-hour cap. +const StaleGrace = time.Minute + // DefaultSTTWindow — how much audio goes to the transcriber in one call. Five // minutes of 16 kHz mono is under 10 MB, transcribes in well under whisper's // own timeout on this box, and keeps the worker responsive to the voice path @@ -87,6 +101,9 @@ var ( // ErrExpired — the session hit MaxDuration and was closed. Returned from // Append so the caller stops sending; the audio collected so far is kept. ErrExpired = errors.New("capture: session reached its time limit") + // ErrWrongSession — the token does not match the running session. The + // recording belongs to the surface that started it. + ErrWrongSession = errors.New("capture: that is not your session") ) // Session — one recording in progress. Not created directly; Recorder.Start @@ -95,11 +112,53 @@ var ( type Session struct { Label string Started time.Time + // Token identifies this session to its owner. Append and Stop need it: the + // rung Append sits on is shared by every writing module, and a rung is not + // an owner. Without it any AuthWrite surface could call capture_stop on a + // meeting it did not start and be handed the verbatim transcript. + Token string mu sync.Mutex - pcm []byte + spool *os.File // the WAV being written, header first + path string + n int64 // PCM bytes written, header excluded format audio.Format expired bool + closed bool +} + +// write appends one frame to the spool file. +func (s *Session) write(b []byte) error { + if s.spool == nil { + return errors.New("capture: session has no spool file") + } + n, err := s.spool.Write(b) + s.n += int64(n) + if err != nil { + return fmt.Errorf("capture: spool write: %w", err) + } + return nil +} + +// finish closes the spool file and stamps the real WAV header over the +// placeholder Start wrote. +func (s *Session) finish() error { + if s.closed { + return nil + } + s.closed = true + if s.spool == nil { + return nil + } + defer s.spool.Close() + hdr, err := audio.WAVHeader(s.format, int(s.n)) + if err != nil { + return err + } + if _, err := s.spool.WriteAt(hdr, 0); err != nil { + return fmt.Errorf("capture: spool header: %w", err) + } + return s.spool.Sync() } // Duration is how much audio has been collected, from the bytes rather than the @@ -112,15 +171,23 @@ func (s *Session) Duration() time.Duration { } func (s *Session) duration() time.Duration { - a := audio.Audio{Format: s.format, Bytes: s.pcm} - return time.Duration(a.Duration() * float64(time.Second)) + return pcmDuration(s.format, s.n) +} + +// pcmDuration is how long n bytes of PCM lasts in the given format. +func pcmDuration(f audio.Format, n int64) time.Duration { + per := int64(f.SampleRate) * int64(f.Channels) * int64(f.SampleBits) / 8 + if per <= 0 { + return 0 + } + return time.Duration(float64(n) / float64(per) * float64(time.Second)) } // Bytes is how much PCM has been collected. For a status line. func (s *Session) Bytes() int { s.mu.Lock() defer s.mu.Unlock() - return len(s.pcm) + return int(s.n) } // Status — what a "что записываешь?" answer needs, and what /dash shows. It is @@ -141,6 +208,7 @@ type Recorder struct { sum *Summarizer maxDuration time.Duration sttWindow time.Duration + staleGrace time.Duration now func() time.Time mu sync.Mutex @@ -180,6 +248,7 @@ func New(blobs *media.Store, tr stt.Transcriber, sum *Summarizer, cfg Config) (* sum: sum, maxDuration: maxDur, sttWindow: window, + staleGrace: StaleGrace, now: time.Now, }, nil } @@ -195,19 +264,84 @@ func (r *Recorder) MaxDuration() time.Duration { return r.maxDuration } func (r *Recorder) Start(label string) (*Session, error) { r.mu.Lock() defer r.mu.Unlock() + r.reapLocked() if r.current != nil { return nil, fmt.Errorf("%w: %q since %s", ErrBusy, r.current.Label, r.current.Started.Format(time.Kitchen)) } + f, err := r.blobs.SpoolFile("capture") + if err != nil { + return nil, err + } + format := audio.PCM16kMono + hdr, err := audio.WAVHeader(format, 0) + if err != nil { + f.Close() + return nil, err + } + // The header is written first and rewritten at Stop with the real length, + // so the spool file is a playable WAV rather than headerless PCM that has + // to be copied to gain 44 bytes. + if _, err := f.Write(hdr); err != nil { + f.Close() + _ = os.Remove(f.Name()) + return nil, fmt.Errorf("capture: spool header: %w", err) + } + token, err := newToken() + if err != nil { + f.Close() + _ = os.Remove(f.Name()) + return nil, err + } s := &Session{ Label: strings.TrimSpace(label), Started: r.now().UTC(), - format: audio.PCM16kMono, + Token: token, + spool: f, + path: f.Name(), + format: format, } r.current = s return s, nil } +// newToken mints a session token. Sixteen random bytes: it is a capability +// handed back over the same socket the call came in on, not a secret at rest. +func newToken() (string, error) { + var b [16]byte + if _, err := rand.Read(b[:]); err != nil { + return "", fmt.Errorf("capture: token: %w", err) + } + return hex.EncodeToString(b[:]), nil +} + +// reapLocked drops a session whose wall clock ran past MaxDuration. The +// frame-driven check in Append only fires while frames arrive, so a client that +// simply stopped sending — a phone whose browser tab was closed, wifi gone — +// left the slot occupied and every later Start answering ErrBusy with a meeting +// from last Tuesday. r.mu must be held. +func (r *Recorder) reapLocked() { + s := r.current + if s == nil { + return + } + if r.now().UTC().Sub(s.Started) < r.maxDuration+r.staleGrace { + return + } + s.mu.Lock() + s.expired = true + _ = s.finish() + path := s.path + s.mu.Unlock() + if path != "" { + // The audio goes with it. A recording nobody stopped is one nobody is + // waiting for, and keeping it would mean storing a meeting on the + // strength of a dropped connection. + _ = os.Remove(path) + } + r.current = nil +} + // Append adds one frame to the running session. ErrNoSession when nothing is // running, which is the guard that makes an ambient path impossible: a stream // arriving at a Recorder nobody started is refused frame by frame. @@ -215,25 +349,39 @@ func (r *Recorder) Start(label string) (*Session, error) { // ErrExpired once the session is at MaxDuration. The audio collected so far is // kept and Stop still works — the cap ends the recording, it does not throw it // away. -func (r *Recorder) Append(a audio.Audio) error { +func (r *Recorder) Append(token string, a audio.Audio) error { if !a.Format.IsValid() { return fmt.Errorf("%w: %+v", ErrBadFormat, a.Format) } r.mu.Lock() + r.reapLocked() s := r.current r.mu.Unlock() if s == nil { return ErrNoSession } + if token != s.Token { + return ErrWrongSession + } s.mu.Lock() defer s.mu.Unlock() if s.expired { return ErrExpired } - s.pcm = append(s.pcm, a.Bytes...) + // The session fixed its format at Start. A client that switches sample rate + // mid-session used to have its frames concatenated into the same buffer: + // duration() then read the whole thing at the original rate, the stored WAV + // header lied, and the cap fired at the wrong length. + if a.Format != s.format { + return fmt.Errorf("%w: session is %+v, frame is %+v", ErrBadFormat, s.format, a.Format) + } + if err := s.write(a.Bytes); err != nil { + return err + } if s.duration() >= r.maxDuration { s.expired = true + _ = s.finish() return ErrExpired } return nil @@ -242,6 +390,7 @@ func (r *Recorder) Append(a audio.Audio) error { // Status reports the running session, or Running=false. func (r *Recorder) Status() Status { r.mu.Lock() + r.reapLocked() s := r.current r.mu.Unlock() if s == nil { @@ -273,6 +422,10 @@ type Result struct { // Chunks — how many windows the transcript was summarised in. 1 means it fit // in one prompt. Reported so a suspiciously vague summary can be explained. Chunks int + // StoreErr — why the audio was not kept, when it was not. The transcript is + // still produced in that case, so this is the difference between "no blob + // because storing failed" and "no blob because nothing was recorded". + StoreErr error } // Stop ends the session and produces the result: store the audio, transcribe it @@ -280,11 +433,18 @@ type Result struct { // the slow work starts, so a stuck model cannot block the next recording. // // The order matters and is the same as vision's: the audio is stored FIRST. If -// transcription or summarisation fails, the recording is still on disk and can -// be run again; a meeting that happened once must not be lost to a model error. -func (r *Recorder) Stop(ctx context.Context) (Result, error) { +// transcription fails, the recording is still on disk under media.retention, so +// the meeting is not lost to a model error. Note that re-running it is a manual +// job today: no method takes a blob id back, unlike vision's Rerun, and the blob +// prunes on the media retention like any other. +func (r *Recorder) Stop(ctx context.Context, token string) (Result, error) { r.mu.Lock() + r.reapLocked() s := r.current + if s != nil && token != s.Token { + r.mu.Unlock() + return Result{}, ErrWrongSession + } r.current = nil r.mu.Unlock() if s == nil { @@ -292,113 +452,168 @@ func (r *Recorder) Stop(ctx context.Context) (Result, error) { } s.mu.Lock() - pcm := s.pcm + err := s.finish() + path := s.path format := s.format + n := s.n s.mu.Unlock() res := Result{Label: s.Label, Started: s.Started} - if len(pcm) == 0 { + if err != nil { + _ = os.Remove(path) + return res, err + } + if n == 0 { + _ = os.Remove(path) return res, ErrEmptyCapture } - full := audio.Audio{Format: format, Bytes: pcm} - res.Duration = time.Duration(full.Duration() * float64(time.Second)) + res.Duration = pcmDuration(format, n) - // Stored as WAV, not headerless PCM: a blob on disk that `aplay` and whisper - // can both open without being told the format is worth 44 bytes. - wav, err := audio.WAVFromPCM(format, pcm) - if err != nil { - return res, fmt.Errorf("capture: wav: %w", err) + // The audio is stored first, as vision does, so a transcription or summary + // failure leaves something to run again. It moves rather than being read + // into memory: a two-hour meeting is a couple of hundred megabytes, and + // this is the process that owns the database and the resident model. + audioPath := path + blob, perr := r.blobs.PutFile(media.KindAudio, "audio/wav", "capture:meeting", path) + if perr == nil { + res.BlobID = blob.ID + audioPath = blob.Path + } else { + // Over the cap, or the store is full. Report it and KEEP GOING: this + // used to return, so the one case the audio cap actually fires on — a + // very long meeting — produced no transcript, no summary and no note, + // which is the whole point of the capability. The spool file stays + // until the transcript has been read off it. + res.StoreErr = perr + defer os.Remove(audioPath) } - blob, err := r.blobs.Put(media.KindAudio, "audio/wav", "capture:meeting", wav) - if err != nil { - // Over the per-blob cap is the expected case for a very long meeting. - // Report it and keep going: a transcript without the audio still beats - // nothing, and the words are what he will read. - return res, fmt.Errorf("capture: store audio: %w", err) - } - res.BlobID = blob.ID - text, err := r.transcribe(ctx, full) - if err != nil { - return res, fmt.Errorf("capture: transcribe: %w", err) - } + text, terr := r.transcribeFile(ctx, audioPath, format, n) res.Transcript = text + if terr != nil { + return res, fmt.Errorf("capture: transcribe: %w", terr) + } if strings.TrimSpace(text) == "" { return res, ErrEmptyCapture } - - if r.sum == nil { - return res, nil + if perr != nil { + return res, fmt.Errorf("capture: store audio: %w", perr) } - summary, chunks, err := r.sum.Summarize(ctx, s.Label, text) - res.Chunks = chunks - if err != nil { - // Degraded success: the transcript is real and stored, only the summary - // is missing. The caller writes the transcript note and says so. - return res, fmt.Errorf("capture: summarize: %w", err) - } - res.Summary = summary return res, nil } +// Summarize runs the map-reduce over a transcript. It is separate from Stop so +// the daemon can answer the stop quickly and do the model work afterwards: a +// full map-reduce is up to forty model calls, and a voice turn that says +// "хватит" should not wait minutes for the reply. +// +// The salvaged text a failed reduce returns is assigned before the error is +// checked. Summarize hands back the per-chunk summaries with its error +// precisely so they are not lost, and the caller used to throw them away. +func (r *Recorder) Summarize(ctx context.Context, res *Result) error { + if r.sum == nil || strings.TrimSpace(res.Transcript) == "" { + return nil + } + summary, chunks, err := r.sum.Summarize(ctx, res.Label, res.Transcript) + res.Chunks = chunks + res.Summary = summary + if err != nil { + return fmt.Errorf("capture: summarize: %w", err) + } + return nil +} + // Abort throws the running session away without transcribing or storing it. // This is what "забудь, не записывай" must map to: a recording someone changed // their mind about leaves nothing behind, not a blob with a note saying it was // abandoned. Returns whether anything was running. -func (r *Recorder) Abort() bool { +func (r *Recorder) Abort(token string) bool { r.mu.Lock() defer r.mu.Unlock() - if r.current == nil { + r.reapLocked() + s := r.current + if s == nil || token != s.Token { return false } r.current = nil + s.mu.Lock() + _ = s.finish() + path := s.path + s.mu.Unlock() + if path != "" { + _ = os.Remove(path) + } return true } -// transcribe runs the transcriber over the audio in windows and joins the text. -// A window that fails is fatal: a summary of a meeting with a silent hole in the -// middle is a summary that misleads. -func (r *Recorder) transcribe(ctx context.Context, a audio.Audio) (string, error) { - windows := chunkAudio(a, r.sttWindow) - parts := make([]string, 0, len(windows)) - for i, w := range windows { - text, _, err := r.tr.Transcribe(ctx, w) +// transcribeFile runs the transcriber over the stored WAV in windows and joins +// the text, reading one window at a time off disk so the meeting is never in +// memory whole. +// +// A window that fails is no longer fatal. It used to be, on the argument that a +// silent hole misleads — but the cost was 24 good windows thrown away for one +// whisper hiccup at minute 100. The hole is marked in the text instead, which +// keeps the words and stays honest about the gap. +func (r *Recorder) transcribeFile(ctx context.Context, path string, format audio.Format, n int64) (string, error) { + f, err := os.Open(path) + if err != nil { + return "", fmt.Errorf("open audio: %w", err) + } + defer f.Close() + + per := windowBytes(format, r.sttWindow) + if per <= 0 || per > n { + per = n + } + total := int((n + per - 1) / per) + buf := make([]byte, per) + parts := make([]string, 0, total) + failed := 0 + for i, off := 0, int64(0); off < n; i, off = i+1, off+per { + size := per + if off+size > n { + size = n - off + } + // Never cut mid-sample: a split inside an int16 shifts every following + // sample by a byte and turns the tail of the window into noise. + if bps := int64(format.SampleBits / 8 * format.Channels); bps > 0 { + size -= size % bps + } + if size <= 0 { + break + } + if _, err := f.ReadAt(buf[:size], int64(audio.WAVHeaderSize)+off); err != nil { + return strings.Join(parts, " "), fmt.Errorf("window %d/%d: %w", i+1, total, err) + } + text, _, err := r.tr.Transcribe(ctx, audio.Audio{Format: format, Bytes: buf[:size]}) if err != nil { - return "", fmt.Errorf("window %d/%d: %w", i+1, len(windows), err) + if ctx.Err() != nil { + return strings.Join(parts, " "), fmt.Errorf("window %d/%d: %w", i+1, total, err) + } + failed++ + parts = append(parts, gapMarker) + continue } if t := strings.TrimSpace(text); t != "" { parts = append(parts, t) } } + if failed == total { + return "", fmt.Errorf("every one of %d window(s) failed", total) + } return strings.Join(parts, " "), nil } -// chunkAudio splits audio into windows of at most window duration, cut on -// sample boundaries. A window shorter than one sample is impossible; audio -// shorter than one window comes back as a single element, so the caller never -// special-cases the short case. -func chunkAudio(a audio.Audio, window time.Duration) []audio.Audio { - bytesPerSample := a.Format.SampleBits / 8 * a.Format.Channels - if bytesPerSample <= 0 || a.Format.SampleRate <= 0 || window <= 0 { - return []audio.Audio{a} +// gapMarker stands in for a window whisper could not read. Russian, because it +// is read by him in a note next to the words around it. +const gapMarker = "[…не разобрала…]" + +// windowBytes is how many PCM bytes one STT window holds. +func windowBytes(f audio.Format, window time.Duration) int64 { + bps := int64(f.SampleBits / 8 * f.Channels) + if bps <= 0 || f.SampleRate <= 0 || window <= 0 { + return 0 } - per := int(window.Seconds()) * a.Format.SampleRate * bytesPerSample - if per <= 0 || len(a.Bytes) <= per { - return []audio.Audio{a} - } - var out []audio.Audio - for off := 0; off < len(a.Bytes); off += per { - end := off + per - if end > len(a.Bytes) { - end = len(a.Bytes) - } - // Never cut mid-sample: a split inside an int16 shifts every following - // sample by a byte and turns the tail of the window into noise. - end -= (end - off) % bytesPerSample - if end <= off { - break - } - out = append(out, audio.Audio{Format: a.Format, Bytes: a.Bytes[off:end]}) - } - return out + per := int64(window.Seconds()) * int64(f.SampleRate) * bps + return per - per%bps } diff --git a/internal/capture/capture_test.go b/internal/capture/capture_test.go index e1c4268..998de21 100644 --- a/internal/capture/capture_test.go +++ b/internal/capture/capture_test.go @@ -90,7 +90,7 @@ func TestNewRequiresStoreAndTranscriber(t *testing.T) { // is refused. There is no ambient path in. func TestAppendWithoutStartIsRefused(t *testing.T) { r, _ := testRecorder(t, &fakeTranscriber{}, nil, Config{}) - if err := r.Append(frame(1)); !errors.Is(err, ErrNoSession) { + if err := r.Append("no-token", frame(1)); !errors.Is(err, ErrNoSession) { t.Fatalf("got %v, want ErrNoSession", err) } if r.Status().Running { @@ -100,20 +100,21 @@ func TestAppendWithoutStartIsRefused(t *testing.T) { func TestStopWithoutStartIsRefused(t *testing.T) { r, _ := testRecorder(t, &fakeTranscriber{}, nil, Config{}) - if _, err := r.Stop(context.Background()); !errors.Is(err, ErrNoSession) { + if _, err := r.Stop(context.Background(), "no-token"); !errors.Is(err, ErrNoSession) { t.Fatalf("got %v, want ErrNoSession", err) } } func TestOneSessionAtATime(t *testing.T) { r, _ := testRecorder(t, &fakeTranscriber{}, nil, Config{}) - if _, err := r.Start("встреча"); err != nil { + s, err := r.Start("встреча") + if err != nil { t.Fatal(err) } if _, err := r.Start("вторая"); !errors.Is(err, ErrBusy) { t.Fatalf("got %v, want ErrBusy", err) } - if _, err := r.Stop(context.Background()); !errors.Is(err, ErrEmptyCapture) { + if _, err := r.Stop(context.Background(), s.Token); !errors.Is(err, ErrEmptyCapture) { t.Fatalf("empty stop: %v", err) } // The slot is free again after a stop, even a failed one. @@ -127,18 +128,22 @@ func TestRoundTripStoresAudioTranscriptAndSummary(t *testing.T) { sum := NewSummarizer(&fakeCompleter{replies: []string{"— решили купить насос"}}, 0, 0, nil) r, blobs := testRecorder(t, tr, sum, Config{}) - if _, err := r.Start("встреча с подрядчиком"); err != nil { + s, err := r.Start("встреча с подрядчиком") + if err != nil { t.Fatal(err) } for i := 0; i < 3; i++ { - if err := r.Append(frame(2)); err != nil { + if err := r.Append(s.Token, frame(2)); err != nil { t.Fatal(err) } } - res, err := r.Stop(context.Background()) + res, err := r.Stop(context.Background(), s.Token) if err != nil { t.Fatalf("stop: %v", err) } + if err := r.Summarize(context.Background(), &res); err != nil { + t.Fatalf("summarize: %v", err) + } if res.BlobID == "" { t.Error("no audio blob stored") } @@ -171,21 +176,22 @@ func TestRoundTripStoresAudioTranscriptAndSummary(t *testing.T) { func TestMaxDurationEndsTheSessionAndKeepsAudio(t *testing.T) { tr := &fakeTranscriber{} r, _ := testRecorder(t, tr, nil, Config{MaxDuration: 4 * time.Second}) - if _, err := r.Start("длинная"); err != nil { + s, err := r.Start("длинная") + if err != nil { t.Fatal(err) } - if err := r.Append(frame(3)); err != nil { + if err := r.Append(s.Token, frame(3)); err != nil { t.Fatalf("first frame: %v", err) } - if err := r.Append(frame(3)); !errors.Is(err, ErrExpired) { + if err := r.Append(s.Token, frame(3)); !errors.Is(err, ErrExpired) { t.Fatalf("got %v, want ErrExpired", err) } // Further frames keep being refused, so a client that ignores the error // cannot grow the recording past the cap. - if err := r.Append(frame(3)); !errors.Is(err, ErrExpired) { + if err := r.Append(s.Token, frame(3)); !errors.Is(err, ErrExpired) { t.Fatalf("post-expiry frame: %v", err) } - res, err := r.Stop(context.Background()) + res, err := r.Stop(context.Background(), s.Token) if err != nil { t.Fatalf("stop after expiry: %v", err) } @@ -196,11 +202,12 @@ func TestMaxDurationEndsTheSessionAndKeepsAudio(t *testing.T) { func TestAppendRejectsWrongFormat(t *testing.T) { r, _ := testRecorder(t, &fakeTranscriber{}, nil, Config{}) - if _, err := r.Start("x"); err != nil { + s, err := r.Start("x") + if err != nil { t.Fatal(err) } bad := audio.Audio{Format: audio.Format{SampleRate: 44100, Channels: 2, SampleBits: 16, Encoding: "pcm_s16le"}, Bytes: make([]byte, 100)} - if err := r.Append(bad); !errors.Is(err, ErrBadFormat) { + if err := r.Append(s.Token, bad); !errors.Is(err, ErrBadFormat) { t.Fatalf("got %v, want ErrBadFormat", err) } } @@ -209,13 +216,14 @@ func TestAppendRejectsWrongFormat(t *testing.T) { func TestAbortLeavesNothing(t *testing.T) { tr := &fakeTranscriber{} r, blobs := testRecorder(t, tr, nil, Config{}) - if _, err := r.Start("зря начали"); err != nil { + s, err := r.Start("зря начали") + if err != nil { t.Fatal(err) } - if err := r.Append(frame(5)); err != nil { + if err := r.Append(s.Token, frame(5)); err != nil { t.Fatal(err) } - if !r.Abort() { + if !r.Abort(s.Token) { t.Fatal("Abort reported nothing running") } if r.Status().Running { @@ -231,7 +239,7 @@ func TestAbortLeavesNothing(t *testing.T) { if tr.calls != 0 { t.Errorf("Abort transcribed anyway (%d calls)", tr.calls) } - if r.Abort() { + if r.Abort(s.Token) { t.Error("second Abort reported a session") } } @@ -241,10 +249,11 @@ func TestStatusReportsTheRunningSession(t *testing.T) { if got := r.Status(); got.Running { t.Error("idle recorder reports running") } - if _, err := r.Start("планёрка"); err != nil { + s, err := r.Start("планёрка") + if err != nil { t.Fatal(err) } - if err := r.Append(frame(10)); err != nil { + if err := r.Append(s.Token, frame(10)); err != nil { t.Fatal(err) } st := r.Status() @@ -264,13 +273,14 @@ func TestStatusReportsTheRunningSession(t *testing.T) { func TestLongAudioIsTranscribedInWindows(t *testing.T) { tr := &fakeTranscriber{} r, _ := testRecorder(t, tr, nil, Config{STTWindow: 2 * time.Second}) - if _, err := r.Start("длинная"); err != nil { + s, err := r.Start("длинная") + if err != nil { t.Fatal(err) } - if err := r.Append(frame(9)); err != nil { + if err := r.Append(s.Token, frame(9)); err != nil { t.Fatal(err) } - res, err := r.Stop(context.Background()) + res, err := r.Stop(context.Background(), s.Token) if err != nil { t.Fatalf("stop: %v", err) } @@ -282,18 +292,19 @@ func TestLongAudioIsTranscribedInWindows(t *testing.T) { } } -// A hole in the middle of a meeting summary would mislead, so a failed window is -// fatal — but the audio is already stored and re-runnable. +// Every window failing is a transcription failure — but the audio is already +// stored and re-runnable. func TestTranscriptionFailureKeepsTheAudio(t *testing.T) { tr := &fakeTranscriber{err: errors.New("whisper is down")} r, blobs := testRecorder(t, tr, nil, Config{}) - if _, err := r.Start("встреча"); err != nil { + s, err := r.Start("встреча") + if err != nil { t.Fatal(err) } - if err := r.Append(frame(2)); err != nil { + if err := r.Append(s.Token, frame(2)); err != nil { t.Fatal(err) } - res, err := r.Stop(context.Background()) + res, err := r.Stop(context.Background(), s.Token) if err == nil { t.Fatal("transcription failure was not reported") } @@ -309,16 +320,20 @@ func TestTranscriptionFailureKeepsTheAudio(t *testing.T) { // error. func TestNoSummarizerStillProducesATranscript(t *testing.T) { r, _ := testRecorder(t, &fakeTranscriber{}, nil, Config{}) - if _, err := r.Start("встреча"); err != nil { + s, err := r.Start("встреча") + if err != nil { t.Fatal(err) } - if err := r.Append(frame(1)); err != nil { + if err := r.Append(s.Token, frame(1)); err != nil { t.Fatal(err) } - res, err := r.Stop(context.Background()) + res, err := r.Stop(context.Background(), s.Token) if err != nil { t.Fatalf("stop: %v", err) } + if err := r.Summarize(context.Background(), &res); err != nil { + t.Fatalf("summarize with no summarizer: %v", err) + } if res.Transcript == "" { t.Error("no transcript") } @@ -331,14 +346,18 @@ func TestNoSummarizerStillProducesATranscript(t *testing.T) { func TestSummaryFailureStillReturnsTheTranscript(t *testing.T) { sum := NewSummarizer(&fakeCompleter{err: errors.New("llama is down")}, 0, 0, nil) r, _ := testRecorder(t, &fakeTranscriber{}, sum, Config{}) - if _, err := r.Start("встреча"); err != nil { + s, err := r.Start("встреча") + if err != nil { t.Fatal(err) } - if err := r.Append(frame(1)); err != nil { + if err := r.Append(s.Token, frame(1)); err != nil { t.Fatal(err) } - res, err := r.Stop(context.Background()) - if err == nil { + res, err := r.Stop(context.Background(), s.Token) + if err != nil { + t.Fatalf("stop: %v", err) + } + if err := r.Summarize(context.Background(), &res); err == nil { t.Fatal("summary failure was not reported") } if res.Transcript == "" { @@ -346,18 +365,12 @@ func TestSummaryFailureStillReturnsTheTranscript(t *testing.T) { } } -func TestChunkAudioNeverCutsMidSample(t *testing.T) { - a := audio.Audio{Format: audio.PCM16kMono, Bytes: make([]byte, 16000*2*5+1)} - for _, w := range chunkAudio(a, 2*time.Second) { - if len(w.Bytes)%2 != 0 { - t.Fatalf("window of %d bytes cuts an int16 in half", len(w.Bytes)) - } - } -} - -func TestChunkAudioShortInputIsOneWindow(t *testing.T) { - a := frame(1) - if got := chunkAudio(a, time.Minute); len(got) != 1 { - t.Errorf("got %d windows, want 1", len(got)) +func TestWindowBytesNeverCutsMidSample(t *testing.T) { + if got := windowBytes(audio.PCM16kMono, 2*time.Second); got%2 != 0 || got != 2*16000*2 { + t.Fatalf("windowBytes = %d", got) + } + odd := audio.Format{SampleRate: 16000, Channels: 1, SampleBits: 16, Encoding: "pcm_s16le"} + if got := windowBytes(odd, 0); got != 0 { + t.Fatalf("a zero window must produce zero, got %d", got) } } diff --git a/internal/capture/session_test.go b/internal/capture/session_test.go new file mode 100644 index 0000000..de31658 --- /dev/null +++ b/internal/capture/session_test.go @@ -0,0 +1,154 @@ +package capture + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/kami/maven/internal/audio" + "github.com/kami/maven/internal/media" +) + +// A frame for a session that already ended must not land in the next one. The +// recorder used to be addressed as "whatever is running now", so a client whose +// session was reaped went on appending its microphone into a meeting somebody +// else had started. +func TestAppendWithTheWrongTokenIsRefused(t *testing.T) { + r, _ := testRecorder(t, &fakeTranscriber{}, nil, Config{}) + s, err := r.Start("первая") + if err != nil { + t.Fatal(err) + } + if err := r.Append("someone-elses-token", frame(1)); !errors.Is(err, ErrWrongSession) { + t.Fatalf("append = %v, want ErrWrongSession", err) + } + if _, err := r.Stop(context.Background(), "someone-elses-token"); !errors.Is(err, ErrWrongSession) { + t.Fatalf("stop = %v, want ErrWrongSession", err) + } + if r.Abort("someone-elses-token") { + t.Fatal("Abort discarded a session it does not own") + } + if err := r.Append(s.Token, frame(1)); err != nil { + t.Fatalf("the owner is still refused: %v", err) + } +} + +// A client that simply stops sending — a phone whose tab was closed — used to +// hold the single session slot forever, and every later Start answered ErrBusy +// with a meeting from last week. +func TestStaleSessionIsReapedByTheWallClock(t *testing.T) { + r, _ := testRecorder(t, &fakeTranscriber{}, nil, Config{MaxDuration: time.Minute}) + now := time.Now().UTC() + r.now = func() time.Time { return now } + s, err := r.Start("брошенная") + if err != nil { + t.Fatal(err) + } + if _, err := r.Start("вторая"); !errors.Is(err, ErrBusy) { + t.Fatalf("start = %v, want ErrBusy", err) + } + now = now.Add(time.Minute + StaleGrace + time.Second) + next, err := r.Start("вторая") + if err != nil { + t.Fatalf("a stale session was not reaped: %v", err) + } + if next.Token == s.Token { + t.Fatal("the new session reused the stale token") + } + if err := r.Append(s.Token, frame(1)); !errors.Is(err, ErrWrongSession) { + t.Fatalf("the reaped client can still write: %v", err) + } + // The abandoned recording is not kept: nobody is waiting for it, and storing + // it would mean keeping a meeting on the strength of a dropped connection. + if _, err := os.Stat(s.path); !os.IsNotExist(err) { + t.Fatalf("the reaped spool file survived: %v", err) + } +} + +// One window failing used to fail the whole transcription, which threw away +// every other window of an hour-long meeting. The hole is marked instead, so the +// summary cannot silently read as if nothing was missing. +func TestOneFailedWindowIsMarkedNotFatal(t *testing.T) { + r, _ := testRecorder(t, &fakeTranscriber{}, nil, Config{STTWindow: time.Second}) + r.tr = &windowTranscriber{failOn: 2} + s, err := r.Start("встреча") + if err != nil { + t.Fatal(err) + } + if err := r.Append(s.Token, frame(3)); err != nil { + t.Fatal(err) + } + res, err := r.Stop(context.Background(), s.Token) + if err != nil { + t.Fatalf("stop: %v", err) + } + if !strings.Contains(res.Transcript, gapMarker) { + t.Errorf("no gap marker in %q", res.Transcript) + } + if !strings.Contains(res.Transcript, "окно1") || !strings.Contains(res.Transcript, "окно3") { + t.Errorf("the surviving windows were dropped: %q", res.Transcript) + } +} + +// The audio not fitting the store is not a reason to lose the words. Stop used +// to return early on a store failure, so a recording over the blob cap produced +// neither a blob nor a transcript. +func TestStoreFailureStillTranscribes(t *testing.T) { + blobs, err := media.OpenWithBudget(t.TempDir(), 512, 1024, 0) + if err != nil { + t.Fatal(err) + } + tr := &fakeTranscriber{} + r, err := New(blobs, tr, nil, Config{}) + if err != nil { + t.Fatal(err) + } + s, err := r.Start("длинная встреча") + if err != nil { + t.Fatal(err) + } + if err := r.Append(s.Token, frame(2)); err != nil { + t.Fatal(err) + } + // The store failure is reported, but as a degraded success: the Result is + // filled in, and the caller keeps it rather than treating the error as + // nothing having happened. + res, err := r.Stop(context.Background(), s.Token) + if err == nil { + t.Fatal("the store failure was not reported") + } + if res.BlobID != "" { + t.Errorf("blob id = %q, want none", res.BlobID) + } + if !errors.Is(res.StoreErr, media.ErrTooLarge) { + t.Errorf("StoreErr = %v, want ErrTooLarge", res.StoreErr) + } + if res.Transcript == "" { + t.Fatal("the words were lost with the audio") + } + // The spool file is cleaned up even on the failure path. + glob, _ := filepath.Glob(filepath.Join(blobs.Dir(), "spool", "*")) + if len(glob) != 0 { + t.Errorf("spool leaked: %v", glob) + } +} + +// windowTranscriber answers per window and fails a chosen one, which is what a +// whisper timeout in the middle of a meeting looks like. +type windowTranscriber struct { + calls int + failOn int +} + +func (w *windowTranscriber) Transcribe(_ context.Context, _ audio.Audio) (string, float64, error) { + w.calls++ + if w.calls == w.failOn { + return "", 0, errors.New("whisper timed out") + } + return fmt.Sprintf("окно%d", w.calls), 1.0, nil +}