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