Maven/cmd/mavend/capture.go

// mavend/capture.go — core's half of the meeting recorder (Vikunja #253,
// docs/plans/08-hearing.md).
//
// The split: a client that has a microphone (mavenclient, or a phone on the PWA)
// is told to start, streams frames over ipc.MethodCaptureAppend, and is told to
// stop. Core keeps the PCM, stores it as a WAV blob under the same media store
// and the same retention as images, transcribes it through the ONE STT Maven has
// (mavsttd's whisper.cpp, reused — not a second engine), and summarises the
// transcript on the resident model in windows that fit n_ctx 4096.
//
// # Off unless configured, twice over
//
// No `media` block ⇒ nowhere to keep audio ⇒ the four capture methods do not
// exist. No `capture` block with enabled ⇒ they still do not exist. On an
// unconfigured box there is no wire path that starts a recording, which is the
// only guarantee worth making about a capability like this one.
//
// # What this file refuses to do
//
//   - Nothing listens. There is no VAD hook here, no wake-word branch, no
//     "start when you hear a meeting". The plan document's keyword-triggered
//     recorder is refused in internal/capture's package comment for the reason
//     that applies here too: noticing a keyword requires listening, which is
//     the behaviour this capability must not have.
//   - No transcript note by default. The summary is written where he will read
//     it; the verbatim record of what other people said takes a deliberate
//     capture.save_transcript.
//   - The transcript is never search input beyond this box, and the audio never
//     leaves it at all.
package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/kami/maven/internal/capture"
	"github.com/kami/maven/internal/config"
	"github.com/kami/maven/internal/ipc"
	"github.com/kami/maven/internal/llm"
	"github.com/kami/maven/internal/phraser"
	"github.com/kami/maven/internal/router"
	"github.com/kami/maven/internal/store"
)

// captureSummaryTimeout — the budget for one summary, which is a map-reduce over
// the whole meeting: one model call per transcript window plus a reduce, each of
// which is seconds on this box. Forty windows is the configured ceiling, so the
// budget has to be minutes, not the 60s the reply path uses. It is spent on a
// background goroutine, never inside the capture_stop request: a client that
// asks Maven to stop recording gets the transcript back in seconds.
const captureSummaryTimeout = 20 * time.Minute

// llmCompleter adapts *llm.Client to capture.Completer. The pure package names
// the two strings it needs and stays free of the llm request struct; the client
// itself is the swap-aware one from llmClientFor, so a model swap re-points it.
type llmCompleter struct {
	c         *llm.Client
	maxTokens int
}

func (l llmCompleter) Complete(ctx context.Context, system, user string) (string, error) {
	return l.c.Complete(ctx, llm.Req{System: system, User: user, MaxTokens: l.maxTokens})
}

// captureWiring — the recorder plus what it needs to write the result down.
type captureWiring struct {
	rec *capture.Recorder
	st  *store.Store
	emb router.Embedder
	cfg *config.CaptureConfig
	now func() time.Time

	// ctx and wg belong to the daemon, not to the request. Summarising happens
	// after the reply has gone out, so it needs a lifetime that outlives the
	// call and a shutdown that waits for it.
	ctx context.Context
	wg  *sync.WaitGroup
}

// newCaptureWiring returns nil when the recorder should not exist: no media
// store, no capture block, capture disabled, or no STT to transcribe with.
//
// A missing llama-server is NOT a reason to return nil. Without one the
// recording is still made, stored and transcribed, and the summary is simply
// absent — the honest degradation, and much better than refusing to record a
// meeting that is happening now.
func newCaptureWiring(ctx context.Context, wg *sync.WaitGroup, keeper *mediaKeeper, st *store.Store, voiceW *voiceWiring, phr phraser.Phraser, emb router.Embedder, cfg *config.Config) *captureWiring {
	if keeper == nil || !cfg.Capture.Records() {
		return nil
	}
	tr := transcriberOf(voiceW)
	if tr == nil {
		// Voice off ⇒ no STT client ⇒ nothing could turn the audio into words.
		// Storing hours of unreadable audio of other people is worse than not
		// recording, so this is a refusal, not a degradation.
		log.Printf("capture: enabled but voice/stt is not wired — meeting capture disabled")
		return nil
	}

	cc := cfg.Capture
	var sum *capture.Summarizer
	if lp, ok := phr.(*phraser.LLMPhraser); ok {
		client := llmClientFor(lp, captureSummaryTimeout)
		sum = capture.NewSummarizer(
			llmCompleter{c: client, maxTokens: 512},
			cc.ChunkRunes, cc.MaxChunks, contextBlockFn(cfg, time.Now),
		)
	} else {
		log.Printf("capture: no llama-server phraser — meetings are transcribed, not summarised")
	}

	rec, err := capture.New(keeper.store, tr, sum, capture.Config{
		MaxDuration: cc.MaxDuration(),
		STTWindow:   time.Duration(cc.STTWindow),
	})
	if err != nil {
		log.Printf("capture: %v — meeting capture disabled", err)
		return nil
	}
	log.Printf("capture: enabled, sessions capped at %s", rec.MaxDuration())
	return &captureWiring{rec: rec, st: st, emb: emb, cfg: cc, now: time.Now, ctx: ctx, wg: wg}
}

// start handles ipc.MethodCaptureStart.
func (c *captureWiring) start(_ context.Context, req ipc.CaptureStartReq) (ipc.CaptureStartResp, error) {
	s, err := c.rec.Start(req.Label)
	if err != nil {
		return ipc.CaptureStartResp{}, err
	}
	// The label is logged; nothing that was said ever is.
	log.Printf("capture: started %q", s.Label)
	return ipc.CaptureStartResp{
		Label:      s.Label,
		Started:    s.Started,
		Token:      s.Token,
		MaxSeconds: int(c.rec.MaxDuration().Seconds()),
	}, nil
}

// append handles ipc.MethodCaptureAppend. ErrExpired is reported as a successful
// response with Expired set rather than an error: the cap firing is the designed
// behaviour, and the client needs the flag to stop sending and call stop.
func (c *captureWiring) append(_ context.Context, req ipc.CaptureAppendReq) (ipc.CaptureAppendResp, error) {
	err := c.rec.Append(req.Token, req.Audio)
	st := c.rec.Status()
	if errors.Is(err, capture.ErrExpired) {
		log.Printf("capture: %q hit the %s cap — stopping", st.Label, c.rec.MaxDuration())
		return ipc.CaptureAppendResp{Seconds: st.Duration.Seconds(), Expired: true}, nil
	}
	if err != nil {
		return ipc.CaptureAppendResp{}, err
	}
	return ipc.CaptureAppendResp{Seconds: st.Duration.Seconds()}, nil
}

// stop handles ipc.MethodCaptureStop.
//
// The error handling here mirrors vision's, and for the same reason: the audio is
// stored first, so a transcription failure returns what exists rather than
// nothing. A response can carry a blob id with no transcript (STT failed,
// re-runnable) — a degraded success, not an error to the caller.
//
// Summarising is NOT done here. A two-hour meeting is forty model calls, which
// on this box is minutes, and holding the IPC request open for them means the
// client that said "стоп" sits there with no answer while its own deadline runs
// out. Stop returns the transcript, and the summary note is written by a
// goroutine in the daemon's WaitGroup afterwards.
func (c *captureWiring) stop(ctx context.Context, req ipc.CaptureStopReq) (ipc.CaptureStopResp, error) {
	if req.Discard {
		// "забудь, не записывай" — nothing is stored, transcribed or noted.
		if !c.rec.Abort(req.Token) {
			return ipc.CaptureStopResp{}, capture.ErrNoSession
		}
		log.Printf("capture: session discarded on request")
		return ipc.CaptureStopResp{Discarded: true}, nil
	}

	res, err := c.rec.Stop(ctx, req.Token)
	resp := ipc.CaptureStopResp{
		BlobID:     res.BlobID,
		Label:      res.Label,
		Started:    res.Started,
		Seconds:    res.Duration.Seconds(),
		Transcript: res.Transcript,
		Summary:    res.Summary,
		Chunks:     res.Chunks,
	}
	if err != nil {
		if res.BlobID == "" && res.Transcript == "" {
			// Nothing survived: no session, or an empty recording. There is
			// nothing to hand back, so this is a real error.
			return ipc.CaptureStopResp{}, err
		}
		log.Printf("capture: %q partially finished: %v", res.Label, err)
	}

	c.summarizeLater(res)
	log.Printf("capture: finished %q — %s of audio, %d bytes of transcript",
		res.Label, res.Duration.Round(time.Second), len(res.Transcript))
	return resp, nil
}

// summarizeLater runs the map-reduce and writes the notes after stop replied.
// The context is the daemon's, not the request's: the request is already
// answered, and cancelling the summary because the client hung up would throw
// away the only readable record of the meeting.
func (c *captureWiring) summarizeLater(res capture.Result) {
	if res.Transcript == "" {
		return
	}
	c.wg.Add(1)
	go func() {
		defer c.wg.Done()
		ctx, cancel := context.WithTimeout(c.ctx, captureSummaryTimeout)
		defer cancel()
		if err := c.rec.Summarize(ctx, &res); err != nil {
			// Not fatal: writeNotes falls back to the transcript, so a dead
			// llama-server costs the summary and not the meeting.
			log.Printf("capture: summary for %q failed: %v", res.Label, err)
		}
		if _, err := c.writeNotes(ctx, res); err != nil {
			log.Printf("capture: note write for %q failed: %v", res.Label, err)
			return
		}
		log.Printf("capture: summarised %q in %d chunk(s)", res.Label, res.Chunks)
	}()
}

// writeNotes stores the summary as a note, and the transcript too when
// capture.save_transcript is set. Returns the id of the note that carries the
// meeting.
//
// With no summary the transcript is written instead, whatever save_transcript
// says. That flag is about keeping the verbatim record IN ADDITION to a summary,
// not about whether the meeting is remembered at all. Without this fallback a
// llama-server that was down at stop time meant an hour of recorded meeting left
// no note behind and nothing recalled it later.
//
// The note source carries the blob id, which is the only link back to the audio.
// When retention prunes the blob the note remains — words about a meeting are a
// far lighter thing to keep than a recording of it.
func (c *captureWiring) writeNotes(ctx context.Context, res capture.Result) (int64, error) {
	source := "capture:meeting"
	if res.BlobID != "" {
		source = "capture:meeting:" + res.BlobID[:12]
	}
	var id int64
	if text := res.Summary; text != "" {
		var err error
		id, err = c.writeNote(ctx, text, source)
		if err != nil {
			return 0, fmt.Errorf("summary note: %w", err)
		}
	} else if res.Transcript != "" {
		var err error
		id, err = c.writeNote(ctx, res.Transcript, source+":transcript")
		if err != nil {
			return 0, fmt.Errorf("transcript note: %w", err)
		}
		return id, nil
	}
	if c.cfg.SaveTranscript && res.Transcript != "" {
		if _, err := c.writeNote(ctx, res.Transcript, source+":transcript"); err != nil {
			return id, fmt.Errorf("transcript note: %w", err)
		}
	}
	return id, nil
}

func (c *captureWiring) writeNote(ctx context.Context, text, source string) (int64, error) {
	var vec []float32
	if c.emb != nil {
		// EmbedPassage, not Embed: this is text being searched FOR, and the e5
		// embedder is asymmetric. Backwards here makes the meeting unfindable by
		// the question that should have matched it.
		var err error
		vec, err = router.EmbedPassage(ctx, c.emb, text)
		if err != nil {
			return 0, fmt.Errorf("embed: %w", err)
		}
	}
	return c.st.WriteNote(ctx, c.now(), text, vec, source)
}

// status handles ipc.MethodCaptureStatus.
func (c *captureWiring) status(_ context.Context) (ipc.CaptureStatusResp, error) {
	st := c.rec.Status()
	return ipc.CaptureStatusResp{
		Running: st.Running,
		Label:   st.Label,
		Started: st.Started,
		Seconds: st.Duration.Seconds(),
		Bytes:   st.Bytes,
	}, nil
}

// wireCapture installs the four IPC hooks, or leaves them nil so every capture
// method reports ErrUnknownMethod. Takes the media keeper wireVision already
// opened: one blob store, one retention loop, images and audio side by side.
func wireCapture(ctx context.Context, wg *sync.WaitGroup, srv *ipc.Server, keeper *mediaKeeper, st *store.Store, voiceW *voiceWiring, phr phraser.Phraser, cfg *config.Config) {
	cw := newCaptureWiring(ctx, wg, keeper, st, voiceW, phr, embedderOf(voiceW), cfg)
	if cw == nil {
		return
	}
	srv.CaptureStartFn = cw.start
	srv.CaptureAppendFn = cw.append
	srv.CaptureStopFn = cw.stop
	srv.CaptureStatusFn = cw.status
}