capture: spool the meeting to disk, own it by token, reap it by the clock

Four invariants the comments claimed and the code did not hold.

The recording lived in mavend's heap as one growing []byte, doubled at
Stop when the WAV was built. Frames now go to a spool file and the
transcript is read back off disk one window at a time, so memory is flat
whatever the length.

A store failure returned before transcription ran, so a meeting over the
blob cap produced no transcript, no summary and no note. It now records
the failure and keeps going, and the spool file survives until the words
have been read off it.

The session had no owner. Any module on the write rung could call stop on
a recording it did not start and receive the verbatim words of everyone
in the room. Start hands back a token and append, stop and abort require
it.

The duration cap was only checked when a frame arrived, so a phone whose
tab was closed left the slot occupied and every later start answered
ErrBusy with a meeting from last week. The wall clock is checked in
start, status, append and stop.

Smaller things in the same pass. Append compares the frame format against
the session format, so a client that switches sample rate mid meeting no
longer has its frames concatenated under a header that lies. One failed
STT window leaves a marker instead of discarding the other twenty four.
Summarize is separate from Stop and assigns the salvaged per chunk text
before it reports the error.

Found in review of #73.

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
 }

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)
 	}
 }

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
+}