diff --git a/cmd/mavend/capture.go b/cmd/mavend/capture.go index de53a70..2393976 100644 --- a/cmd/mavend/capture.go +++ b/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 } diff --git a/cmd/mavend/capture_test.go b/cmd/mavend/capture_test.go new file mode 100644 index 0000000..e09bdad --- /dev/null +++ b/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) + } +} diff --git a/cmd/mavend/main.go b/cmd/mavend/main.go index 8211d0e..9e5ac68 100644 --- a/cmd/mavend/main.go +++ b/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. diff --git a/internal/config/config.go b/internal/config/config.go index 4fdae93..b9a4c26 100644 --- a/internal/config/config.go +++ b/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"` } diff --git a/internal/config/senses_test.go b/internal/config/senses_test.go index 8f50a17..44ec12c 100644 --- a/internal/config/senses_test.go +++ b/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) { diff --git a/internal/ipc/api.go b/internal/ipc/api.go index c2ff155..0083ae6 100644 --- a/internal/ipc/api.go +++ b/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"` diff --git a/internal/ipc/client.go b/internal/ipc/client.go index fae2f4d..fa9852e 100644 --- a/internal/ipc/client.go +++ b/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 {