cmd/mavend/capture.go

@@ -34,6 +34,7 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"sync"
 	"time"
 
 	"github.com/kami/maven/internal/capture"
@@ -45,10 +46,12 @@ import (
 	"github.com/kami/maven/internal/store"
 )
 
-// captureSummaryTimeout — the budget for one Stop, which is a map-reduce over
+// 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.
+// 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
@@ -70,6 +73,12 @@ type captureWiring struct {
 	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
@@ -79,7 +88,7 @@ type captureWiring struct {
 // 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(keeper *mediaKeeper, st *store.Store, voiceW *voiceWiring, phr phraser.Phraser, emb router.Embedder, cfg *config.Config) *captureWiring {
+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
 	}
@@ -113,7 +122,7 @@ func newCaptureWiring(keeper *mediaKeeper, st *store.Store, voiceW *voiceWiring,
 		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}
+	return &captureWiring{rec: rec, st: st, emb: emb, cfg: cc, now: time.Now, ctx: ctx, wg: wg}
 }
 
 // start handles ipc.MethodCaptureStart.
@@ -127,6 +136,7 @@ func (c *captureWiring) start(_ context.Context, req ipc.CaptureStartReq) (ipc.C
 	return ipc.CaptureStartResp{
 		Label:      s.Label,
 		Started:    s.Started,
+		Token:      s.Token,
 		MaxSeconds: int(c.rec.MaxDuration().Seconds()),
 	}, nil
 }
@@ -135,7 +145,7 @@ func (c *captureWiring) start(_ context.Context, req ipc.CaptureStartReq) (ipc.C
 // 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.Audio)
+	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())
@@ -150,21 +160,26 @@ func (c *captureWiring) append(_ context.Context, req ipc.CaptureAppendReq) (ipc
 // stop handles ipc.MethodCaptureStop.
 //
 // The error handling here mirrors vision's, and for the same reason: the audio is
-// stored first, so a transcription or summary failure returns what exists rather
-// than nothing. A response can carry a blob id with no transcript (STT failed,
-// re-runnable), or a transcript with no summary (the model failed, the words are
-// kept) — both are degraded successes and neither is an error to the caller.
+// 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() {
+		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)
+	res, err := c.rec.Stop(ctx, req.Token)
 	resp := ipc.CaptureStopResp{
 		BlobID:     res.BlobID,
 		Label:      res.Label,
@@ -183,19 +198,47 @@ func (c *captureWiring) stop(ctx context.Context, req ipc.CaptureStopReq) (ipc.C
 		log.Printf("capture: %q partially finished: %v", res.Label, err)
 	}
 
-	if id, werr := c.writeNotes(ctx, res); werr != nil {
-		log.Printf("capture: note write for %q failed: %v", res.Label, werr)
-	} else {
-		resp.NoteID = id
-	}
-	log.Printf("capture: finished %q — %s of audio, %d summary chunk(s)",
-		res.Label, res.Duration.Round(time.Second), res.Chunks)
+	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 summary note's id, or 0 when there
-// was no summary to write.
+// 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
@@ -212,6 +255,13 @@ func (c *captureWiring) writeNotes(ctx context.Context, res capture.Result) (int
 		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 {
@@ -251,8 +301,8 @@ func (c *captureWiring) status(_ context.Context) (ipc.CaptureStatusResp, error)
 // 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(srv *ipc.Server, keeper *mediaKeeper, st *store.Store, voiceW *voiceWiring, phr phraser.Phraser, cfg *config.Config) {
-	cw := newCaptureWiring(keeper, st, voiceW, phr, embedderOf(voiceW), cfg)
+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
 	}

cmd/mavend/capture_test.go

@@ -0,0 +1,118 @@
+package main
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/kami/maven/internal/audio"
+	"github.com/kami/maven/internal/capture"
+	"github.com/kami/maven/internal/config"
+	"github.com/kami/maven/internal/ipc"
+	"github.com/kami/maven/internal/media"
+)
+
+// silentTranscriber stands in for mavsttd: one fixed phrase per window, so the
+// wiring can be tested without whisper.
+type silentTranscriber struct{}
+
+func (silentTranscriber) Transcribe(_ context.Context, _ audio.Audio) (string, float64, error) {
+	return "решили купить насос", 1.0, nil
+}
+
+func testCaptureWiring(t *testing.T) (*captureWiring, *sync.WaitGroup) {
+	t.Helper()
+	blobs, err := media.Open(t.TempDir(), 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rec, err := capture.New(blobs, silentTranscriber{}, nil, capture.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var wg sync.WaitGroup
+	return &captureWiring{
+		rec: rec,
+		st:  newTestStore(t),
+		cfg: &config.CaptureConfig{},
+		now: time.Now,
+		ctx: context.Background(),
+		wg:  &wg,
+	}, &wg
+}
+
+// A frame carrying the wrong token must not land in the running session. Append
+// and stop used to address "whatever is running now", so a client whose session
+// had already ended went on recording into somebody else's meeting, and any
+// client could end a recording it never started.
+func TestCaptureRefusesAnotherClientsToken(t *testing.T) {
+	c, _ := testCaptureWiring(t)
+	start, err := c.start(context.Background(), ipc.CaptureStartReq{Label: "встреча"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if start.Token == "" {
+		t.Fatal("start handed back no session token")
+	}
+	if _, err := c.append(context.Background(), ipc.CaptureAppendReq{
+		Token: "not-mine",
+		Audio: audio.Audio{Format: audio.PCM16kMono, Bytes: make([]byte, 3200)},
+	}); err == nil {
+		t.Error("a frame with the wrong token was accepted")
+	}
+	if _, err := c.stop(context.Background(), ipc.CaptureStopReq{Token: "not-mine"}); err == nil {
+		t.Error("a stop with the wrong token ended the session")
+	}
+	if st, _ := c.status(context.Background()); !st.Running {
+		t.Error("the session was ended by a client that does not own it")
+	}
+}
+
+// Stop answers with the transcript and does not wait for the summary. The
+// summary is up to forty model calls, and holding the IPC request for them meant
+// the client that said "стоп" sat with no answer for minutes.
+//
+// With no summariser wired the note still has to be written, from the transcript.
+// save_transcript is about keeping the verbatim record IN ADDITION to a summary,
+// not about whether the meeting is remembered at all — without this fallback a
+// dead llama-server meant an hour of meeting left no note behind.
+func TestStopReturnsTranscriptAndNotesItWithoutASummary(t *testing.T) {
+	c, wg := testCaptureWiring(t)
+	start, err := c.start(context.Background(), ipc.CaptureStartReq{Label: "планёрка"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := c.append(context.Background(), ipc.CaptureAppendReq{
+		Token: start.Token,
+		Audio: audio.Audio{Format: audio.PCM16kMono, Bytes: make([]byte, 32000)},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	resp, err := c.stop(context.Background(), ipc.CaptureStopReq{Token: start.Token})
+	if err != nil {
+		t.Fatalf("stop: %v", err)
+	}
+	if resp.Transcript == "" {
+		t.Fatal("stop returned no transcript")
+	}
+	if resp.Summary != "" {
+		t.Errorf("summary = %q, want none inside the request", resp.Summary)
+	}
+	wg.Wait()
+
+	notes, err := c.st.RecentNotes(context.Background(), 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var found bool
+	for _, n := range notes {
+		if strings.Contains(n.Text, "насос") {
+			found = true
+		}
+	}
+	if !found {
+		t.Fatalf("the meeting left no note behind: %+v", notes)
+	}
+}

cmd/mavend/main.go

@@ -385,7 +385,7 @@ func run(args []string) error {
 		// The meeting recorder (Vikunja #253) shares that blob store and its
 		// retention loop. Off unless a capture block enables it, in which case
 		// all four capture methods answer ErrUnknownMethod.
-		wireCapture(srv, keeper, st, voiceW, phr, cfg)
+		wireCapture(ctx, &wg, srv, keeper, st, voiceW, phr, cfg)
 		// Voice identification (Vikunja #255). Enrolment plumbing only until a
 		// speaker-embedding model exists on disk; off entirely without a speaker
 		// block, so no wire path takes a voiceprint on a default box.
@@ -561,7 +561,7 @@ func run(args []string) error {
 			wireMailIntake(srv, st, phr, cfg, evBus)
 			wireModelSwap(srv, phr, cfg)
 			keeper := wireVision(ctx, &wg, srv, st, embedderOf(voiceW), cfg)
-			wireCapture(srv, keeper, st, voiceW, phr, cfg)
+			wireCapture(ctx, &wg, srv, keeper, st, voiceW, phr, cfg)
 			// Voice identification (Vikunja #255). Enrolment plumbing only until a
 			// speaker-embedding model exists on disk; off entirely without a speaker
 			// block, so no wire path takes a voiceprint on a default box.

internal/config/config.go

@@ -27,8 +27,8 @@ import (
 	"github.com/kami/maven/internal/morning"
 	"github.com/kami/maven/internal/netscan"
 	"github.com/kami/maven/internal/smarthome"
-	"github.com/kami/maven/internal/vision"
 	"github.com/kami/maven/internal/update"
+	"github.com/kami/maven/internal/vision"
 	"github.com/robfig/cron/v3"
 )
 
@@ -787,9 +787,14 @@ type CaptureConfig struct {
 	MaxChunks int `json:"max_chunks,omitempty"`
 
 	// SaveTranscript — write the full transcript as a note alongside the
-	// summary. Default false: a verbatim record of what other people said in a
-	// room is a heavier thing to keep than a four-line summary, so it takes a
-	// deliberate yes. The audio blob is pruned by media.retention either way.
+	// summary. Default false, and the cost is not disk: a note is embedded and
+	// becomes recall corpus, so every later question can surface verbatim words
+	// other people said in a room. That is the reason it takes a deliberate yes.
+	// The audio blob is pruned by media.retention either way; the notes are not.
+	//
+	// A meeting with no summary writes its transcript regardless. The choice
+	// here is transcript IN ADDITION to a summary, not whether the meeting is
+	// remembered at all.
 	SaveTranscript bool `json:"save_transcript,omitempty"`
 }
 

internal/config/senses_test.go

@@ -228,14 +228,14 @@ func TestSpeakerBlockParsesFromJSON(t *testing.T) {
 // both fail at startup now.
 func TestSensesBlocksAreValidatedAtStartup(t *testing.T) {
 	bad := map[string]string{
-		"media with no dir":     `{"media":{"retention":"48h"}}`,
-		"negative budget":       `{"media":{"dir":"/srv/media","max_total_bytes":-1}}`,
-		"blob over the budget":  `{"media":{"dir":"/srv/media","max_bytes":100,"max_total_bytes":10}}`,
-		"vision with no media dir":  `{"vision":{"enabled":true,"endpoint":"http://127.0.0.1:8081"}}`,
-		"vision endpoint typo":  `{"media":{"dir":"/srv/media"},"vision":{"enabled":true,"endpoint":"127.0.0.1:8081"}}`,
-		"vision on the wan":   `{"media":{"dir":"/srv/media"},"vision":{"enabled":true,"endpoint":"http://8.8.8.8:8081"}}`,
-		"vision, empty endpoint": `{"media":{"dir":"/srv/media"},"vision":{"enabled":true}}`,
-		"capture with no store":  `{"capture":{"enabled":true}}`,
+		"media with no dir":        `{"media":{"retention":"48h"}}`,
+		"negative budget":          `{"media":{"dir":"/srv/media","max_total_bytes":-1}}`,
+		"blob over the budget":     `{"media":{"dir":"/srv/media","max_bytes":100,"max_total_bytes":10}}`,
+		"vision with no media dir": `{"vision":{"enabled":true,"endpoint":"http://127.0.0.1:8081"}}`,
+		"vision endpoint typo":     `{"media":{"dir":"/srv/media"},"vision":{"enabled":true,"endpoint":"127.0.0.1:8081"}}`,
+		"vision on the wan":        `{"media":{"dir":"/srv/media"},"vision":{"enabled":true,"endpoint":"http://8.8.8.8:8081"}}`,
+		"vision, empty endpoint":   `{"media":{"dir":"/srv/media"},"vision":{"enabled":true}}`,
+		"capture with no store":    `{"capture":{"enabled":true}}`,
 	}
 	for name, body := range bad {
 		t.Run(name, func(t *testing.T) {

internal/ipc/api.go

@@ -247,9 +247,14 @@ type CaptureStartReq struct {
 // which it stops itself; the caller tells him, so a forgotten recording is his
 // own informed choice rather than a surprise.
 type CaptureStartResp struct {
-	Label      string    `json:"label,omitempty"`
-	Started    time.Time `json:"started"`
-	MaxSeconds int       `json:"max_seconds"`
+	Label   string    `json:"label,omitempty"`
+	Started time.Time `json:"started"`
+	// Token names THIS session. Every later append, stop and discard has to
+	// carry it. Without it the recorder is addressed by "whatever is running
+	// now", and a client whose session already ended on the duration cap goes on
+	// appending its microphone into the next session someone else started.
+	Token      string `json:"token"`
+	MaxSeconds int    `json:"max_seconds"`
 }
 
 // CaptureAppendReq — one chunk of audio for the running session. Refused with
@@ -257,6 +262,9 @@ type CaptureStartResp struct {
 // makes an ambient path impossible: audio arriving at an idle core is dropped on
 // the floor, not buffered "just in case".
 type CaptureAppendReq struct {
+	// Token from CaptureStartResp. A frame for a session that already ended is
+	// refused rather than folded into whatever is running now.
+	Token string      `json:"token"`
 	Audio audio.Audio `json:"audio"`
 }
 
@@ -275,15 +283,22 @@ type CaptureAppendResp struct {
 // flag rather than a separate method so the client that says "stop" and the
 // client that says "stop and forget" take the same path to the same session.
 type CaptureStopReq struct {
-	Discard bool `json:"discard,omitempty"`
+	// Token from CaptureStartResp. Stopping by "whatever is running" lets a
+	// late client end a recording it never started.
+	Token   string `json:"token"`
+	Discard bool   `json:"discard,omitempty"`
 }
 
 // CaptureStopResp — the finished capture. BlobID is the stored WAV, kept under
 // media.retention like any other blob and pruned with it.
 //
-// A response with a Transcript and an empty Summary is a degraded success: the
-// words exist, only the model failed. A response with a BlobID and neither is
-// the audio surviving a transcription failure — the same id can be run again.
+// A response with a Transcript and an empty Summary is the normal shape, not a
+// failure: summarising a long meeting is a map-reduce of minutes, so stop
+// answers with the words and the summary note is written afterwards. Summary is
+// filled in only when it happened to be ready. A response with a BlobID and no
+// transcript is the audio surviving a transcription failure — the same id can be
+// run again by hand off the blob before media.retention prunes it — there is no
+// capture method that takes a blob id, so this is not a re-run the wire offers.
 // Discarded is true when nothing was kept.
 type CaptureStopResp struct {
 	BlobID     string    `json:"blob_id,omitempty"`

internal/ipc/client.go

@@ -500,9 +500,10 @@ func (c *Client) CaptureAppend(ctx context.Context, req CaptureAppendReq) (Captu
 	return r, nil
 }
 
-// CaptureStop ends the session. Slow — it transcribes and summarises the whole
-// recording — so pass a context with room. Set Discard to throw the recording
-// away instead.
+// CaptureStop ends the session. It transcribes the whole recording before
+// answering, so pass a context with room; the summary is written afterwards by
+// the daemon and is usually absent from the response. Set Discard to throw the
+// recording away instead. Token comes from CaptureStart.
 func (c *Client) CaptureStop(ctx context.Context, req CaptureStopReq) (CaptureStopResp, error) {
 	var r CaptureStopResp
 	if err := c.call(ctx, MethodCaptureStop, req, &r); err != nil {