Maven/internal/capture/capture.go

// Package capture is Maven's meeting recorder (Vikunja #253,
// docs/plans/08-hearing.md).
//
// One session at a time, with an explicit start and an explicit stop:
//
//	Start("встреча")  → audio frames appended → Stop() → transcript → summary
//
// # Nothing here listens
//
// This is the most invasive capability in the backlog and the design is
// constrained accordingly. The constraints are the code, not a preamble:
//
//   - There is no ambient path. `Session.Append` is the only way audio enters,
//     and it only accepts frames while a session someone started is running.
//     A keyword-triggered recorder ("maven record" heard in the room) was in the
//     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 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.
//   - The transcript is never search input for anything outside this box. It is
//     text about a conversation with other people in it.
//
// # Long audio against a 4096-token context
//
// The resident model is a Thinking variant at n_ctx 4096, so an hour of meeting
// transcript does not fit in one prompt and never will. summarize.go does the
// obvious map-reduce: split the transcript on sentence boundaries into windows
// that fit, summarise each, then summarise the summaries. That is handled
// explicitly rather than by truncation, because a truncated meeting summary is
// worse than none — it looks complete and is not.
//
// # Transcription
//
// 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 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. 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"

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

// 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 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
// between windows.
const DefaultSTTWindow = 5 * time.Minute

// Errors callers distinguish.
var (
	// ErrDisabled — capture is not configured. A capability is off unless
	// configured, and a recorder most of all.
	ErrDisabled = errors.New("capture: not configured")
	// ErrBusy — a session is already running. One at a time: two concurrent
	// recordings would make "хватит" ambiguous.
	ErrBusy = errors.New("capture: a session is already running")
	// ErrNoSession — stop or append with nothing running.
	ErrNoSession = errors.New("capture: nothing is being recorded")
	// ErrBadFormat — a frame is not the canonical 16 kHz mono PCM shape.
	ErrBadFormat = errors.New("capture: audio format not supported")
	// ErrEmptyCapture — the session ended with no audio in it.
	ErrEmptyCapture = errors.New("capture: nothing was recorded")
	// 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
// makes it. Guarded by a mutex because frames arrive from a network goroutine
// while a status call may read from another.
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
	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
// wall clock: a stream that dropped frames should report the audio that exists,
// not the time that passed.
func (s *Session) Duration() time.Duration {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.duration()
}

func (s *Session) duration() time.Duration {
	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 int(s.n)
}

// Status — what a "что записываешь?" answer needs, and what /dash shows. It is
// the read side of a running session and is safe to ask for at any time.
type Status struct {
	Running  bool          `json:"running"`
	Label    string        `json:"label,omitempty"`
	Started  time.Time     `json:"started,omitempty"`
	Duration time.Duration `json:"duration,omitempty"`
	Bytes    int           `json:"bytes,omitempty"`
}

// Recorder owns the single session slot, the blob store and the two models a
// finished capture needs. Build it with New; a zero Recorder is not usable.
type Recorder struct {
	blobs       *media.Store
	tr          stt.Transcriber
	sum         *Summarizer
	maxDuration time.Duration
	sttWindow   time.Duration
	staleGrace  time.Duration
	now         func() time.Time

	mu      sync.Mutex
	current *Session
}

// Config — the recorder's knobs, built from config.CaptureConfig by the daemon.
type Config struct {
	// MaxDuration — hard cap on one session. 0 ⇒ DefaultMaxDuration.
	MaxDuration time.Duration
	// STTWindow — audio per transcription call. 0 ⇒ DefaultSTTWindow.
	STTWindow time.Duration
}

// New builds a Recorder. blobs and tr are required — a recorder with nowhere to
// put the audio, or nothing to transcribe it with, is not a recorder. sum may be
// nil: the transcript is still produced and stored, and the summary is simply
// absent, which is the honest degradation when there is no llama-server.
func New(blobs *media.Store, tr stt.Transcriber, sum *Summarizer, cfg Config) (*Recorder, error) {
	if blobs == nil {
		return nil, errors.New("capture: no blob store")
	}
	if tr == nil {
		return nil, errors.New("capture: no transcriber")
	}
	maxDur := cfg.MaxDuration
	if maxDur <= 0 {
		maxDur = DefaultMaxDuration
	}
	window := cfg.STTWindow
	if window <= 0 {
		window = DefaultSTTWindow
	}
	return &Recorder{
		blobs:       blobs,
		tr:          tr,
		sum:         sum,
		maxDuration: maxDur,
		sttWindow:   window,
		staleGrace:  StaleGrace,
		now:         time.Now,
	}, nil
}

// MaxDuration is the configured hard cap. For the reply that tells him how long
// she will keep going if he forgets to say "хватит".
func (r *Recorder) MaxDuration() time.Duration { return r.maxDuration }

// Start opens a session. label is what the meeting is called ("встреча с
// подрядчиком"); it ends up in the summary note so the note is findable.
// ErrBusy if one is already running — the caller says so rather than silently
// discarding the first recording.
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(),
		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.
//
// 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(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
	}
	// 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
}

// 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 {
		return Status{}
	}
	return Status{
		Running:  true,
		Label:    s.Label,
		Started:  s.Started,
		Duration: s.Duration(),
		Bytes:    s.Bytes(),
	}
}

// Result — a finished capture.
type Result struct {
	// BlobID — the stored audio, content-addressed. Empty only if storing failed.
	BlobID string
	// Label / Started / Duration — what was recorded and when.
	Label    string
	Started  time.Time
	Duration time.Duration
	// Transcript — the full text, joined across STT windows.
	Transcript string
	// Summary — the map-reduced summary, or empty when no summarizer was wired
	// or the model failed. Empty summary with a non-empty transcript is a
	// degraded success, not a failure: the words are there.
	Summary string
	// 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
// in windows, summarise it in windows. The session slot is freed before any of
// 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 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 {
		return Result{}, ErrNoSession
	}

	s.mu.Lock()
	err := s.finish()
	path := s.path
	format := s.format
	n := s.n
	s.mu.Unlock()

	res := Result{Label: s.Label, Started: s.Started}
	if err != nil {
		_ = os.Remove(path)
		return res, err
	}
	if n == 0 {
		_ = os.Remove(path)
		return res, ErrEmptyCapture
	}
	res.Duration = pcmDuration(format, n)

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

	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 perr != nil {
		return res, fmt.Errorf("capture: store audio: %w", perr)
	}
	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(token string) bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	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
}

// 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 {
			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
}

// 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 := int64(window.Seconds()) * int64(f.SampleRate) * bps
	return per - per%bps
}