cmd/mavend/mail.go

@@ -22,7 +22,9 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"strings"
 	"time"
+	"unicode"
 
 	"github.com/kami/maven/internal/config"
 	"github.com/kami/maven/internal/email"
@@ -37,6 +39,35 @@ import (
 // list into a copy of his mailbox.
 const evidenceMaxChars = 160
 
+// captureTimeout — how long the capture writes get, separately from the
+// extraction budget. A candidate the model already produced must not be lost
+// because the model was slow.
+const captureTimeout = 30 * time.Second
+
+// maxMailboxChars — a mailbox name is an IMAP folder, not free text. It ends up
+// in the provenance string, which is a small controlled vocabulary.
+const maxMailboxChars = 64
+
+// validMailbox checks the name this method is willing to write provenance for.
+// Empty is refused: "email:" is not a source. So is anything with a control
+// character or a space-only value, so the source string stays greppable and
+// stays one token.
+func validMailbox(s string) (string, error) {
+	s = strings.TrimSpace(s)
+	if s == "" {
+		return "", fmt.Errorf("mail intake: mailbox is required")
+	}
+	if len([]rune(s)) > maxMailboxChars {
+		return "", fmt.Errorf("mail intake: mailbox name too long")
+	}
+	for _, r := range s {
+		if r < 0x20 || r == 0x7f || unicode.IsSpace(r) {
+			return "", fmt.Errorf("mail intake: mailbox name has whitespace or a control character")
+		}
+	}
+	return s, nil
+}
+
 // mailIntake — extraction + capture for one message at a time.
 type mailIntake struct {
 	st      *store.Store
@@ -64,15 +95,25 @@ func newMailIntake(st *store.Store, phr phraser.Phraser, cfg *config.Config, bus
 	}
 	lp, ok := phr.(*phraser.LLMPhraser)
 	if !ok {
-		log.Printf("mail intake: configured but no llama-server phraser — mail ingestion disabled")
+		// The phraser is not an *LLMPhraser. Today that means there is no
+		// llama-server; if anything ever WRAPS the phraser it will mean that
+		// instead, so the line names the assertion rather than guessing why.
+		log.Printf("mail intake: configured but the phraser is not an *phraser.LLMPhraser (%T) — mail ingestion disabled", phr)
 		return nil
 	}
 	timeout := time.Duration(cfg.Email.Timeout)
 	if timeout <= 0 {
 		timeout = config.DefaultEmailTimeout
 	}
-	ex := email.NewExtractor(llmClientFor(lp, timeout), cfg.Email.MaxTasks, contextBlockFn(cfg, time.Now))
-	log.Printf("mail intake: enabled (max %d candidates per message, timeout %s)", cfg.Email.MaxTasks, timeout)
+	// Background client: extraction is a job nobody is waiting on, and it shares
+	// one llama-server slot with the voice turn. Through the gate it yields to
+	// anything he is waiting for and only one extraction runs at a time, so a
+	// first poll of 25 unseen messages cannot queue 25 model calls in front of
+	// him. See llm.Gate.
+	ex := email.NewExtractor(llmBackgroundClientFor(lp, timeout), cfg.Email.MaxTasks, contextBlockFn(cfg, time.Now))
+	// The NORMALISED bound, not the configured one: with "email": {} in
+	// mavend.json the configured value is 0 and the daemon allows three.
+	log.Printf("mail intake: enabled (max %d candidates per message, timeout %s)", ex.Max(), timeout)
 	return &mailIntake{st: st, ex: ex, timeout: timeout, now: time.Now, bus: bus}
 }
 
@@ -86,6 +127,13 @@ func newMailIntake(st *store.Store, phr phraser.Phraser, cfg *config.Config, bus
 // live rows, so a mailbox re-read after a restart produces Created=0 rather
 // than a second copy of every task.
 func (m *mailIntake) ingest(ctx context.Context, req ipc.IngestMailReq) (ipc.IngestMailResp, error) {
+	// The mailbox name becomes provenance ("email:INBOX"), and the source
+	// vocabulary is what the loop's rules trust. An empty name gave "email:" and
+	// an arbitrary string gave an arbitrary source under that namespace.
+	mailbox, err := validMailbox(req.Mailbox)
+	if err != nil {
+		return ipc.IngestMailResp{}, err
+	}
 	msg := email.Message{
 		UID:     req.UID,
 		From:    req.From,
@@ -98,9 +146,15 @@ func (m *mailIntake) ingest(ctx context.Context, req ipc.IngestMailReq) (ipc.Ing
 		return ipc.IngestMailResp{Skipped: true}, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, m.timeout)
-	defer cancel()
-	cands, err := m.ex.Extract(ctx, msg)
+	// The timeout scopes the EXTRACTION and nothing else. It used to wrap the
+	// capture writes too, so a model that answered at 119 seconds of a 120
+	// second budget left the first CaptureTask one second and the third none:
+	// the work was done, the answer was good, and it was dropped with a
+	// deadline error. Config calls this a per-message extraction budget, and now
+	// it is one.
+	exCtx, cancel := context.WithTimeout(ctx, m.timeout)
+	cands, err := m.ex.Extract(exCtx, msg)
+	cancel()
 	if err != nil {
 		// The error from internal/email never carries mail text; keep it that way
 		// by not adding the subject here.
@@ -110,7 +164,13 @@ func (m *mailIntake) ingest(ctx context.Context, req ipc.IngestMailReq) (ipc.Ing
 		return ipc.IngestMailResp{}, nil
 	}
 
-	source := email.SourcePrefix + req.Mailbox
+	// A fresh budget for the writes, derived from the caller's context rather
+	// than from the extraction's. Encrypted-store writes are fast; what this
+	// bounds is a stuck store, not the model.
+	ctx, cancel = context.WithTimeout(ctx, captureTimeout)
+	defer cancel()
+
+	source := email.SourcePrefix + mailbox
 	evidence := truncateRunes(req.Subject, evidenceMaxChars)
 	now := m.now()
 	var resp ipc.IngestMailResp

cmd/mavend/mail_test.go

@@ -180,3 +180,62 @@ func TestNewMailIntakeOffWithoutConfig(t *testing.T) {
 		t.Error("without a llama-server phraser there is nothing to extract with")
 	}
 }
+
+// The mailbox name becomes the provenance string, which is the vocabulary the
+// loop's rules trust. "email:" is not a source and neither is "email:anything
+// he could post at the socket".
+func TestIngestRejectsBadMailbox(t *testing.T) {
+	for _, name := range []string{"", "   ", "IN BOX", "IN\nBOX", "IN\x00BOX", strings.Repeat("щ", maxMailboxChars+1)} {
+		mi, st, fake := newTestIntake(t, `[{"text":"дело","due":""}]`)
+		req := ingestReq()
+		req.Mailbox = name
+		if _, err := mi.ingest(context.Background(), req); err == nil {
+			t.Errorf("mailbox %q was accepted", name)
+		}
+		if fake.calls != 0 {
+			t.Errorf("mailbox %q reached the model", name)
+		}
+		if tasks, _ := st.ListTasks(context.Background(), ""); len(tasks) != 0 {
+			t.Errorf("mailbox %q wrote %d tasks", name, len(tasks))
+		}
+	}
+}
+
+// slowLLM burns most of the extraction budget before answering, the way a
+// Thinking 1.7B does on a long mail.
+type slowLLM struct {
+	reply string
+	delay time.Duration
+}
+
+func (s *slowLLM) Complete(ctx context.Context, _ llm.Req) (string, error) {
+	select {
+	case <-time.After(s.delay):
+		return s.reply, nil
+	case <-ctx.Done():
+		return "", ctx.Err()
+	}
+}
+
+// The extraction budget must not also bound the writes. It used to be one
+// context, so a model answering near the deadline lost the candidates it had
+// just produced.
+func TestIngestCapturesAfterASlowExtraction(t *testing.T) {
+	st := newTestStore(t)
+	mi := &mailIntake{
+		st:      st,
+		ex:      email.NewExtractor(&slowLLM{reply: `[{"text":"оплатить счёт","due":""}]`, delay: 90 * time.Millisecond}, 0, nil),
+		timeout: 100 * time.Millisecond,
+		now:     func() time.Time { return time.Date(2026, 8, 1, 10, 0, 0, 0, time.UTC) },
+	}
+	resp, err := mi.ingest(context.Background(), ingestReq())
+	if err != nil {
+		t.Fatalf("ingest: %v", err)
+	}
+	if resp.Created != 1 {
+		t.Fatalf("resp = %+v, want the candidate captured", resp)
+	}
+	if tasks, _ := st.ListTasks(context.Background(), ""); len(tasks) != 1 {
+		t.Errorf("got %d tasks, want 1", len(tasks))
+	}
+}

cmd/mavend/memoryeval.go

@@ -49,7 +49,9 @@ func newMemoryEvalWorker(st *store.Store, phr phraser.Phraser, cfg *config.Confi
 	}
 	// A generous per-request timeout: this is a long prompt to a Thinking model
 	// and nobody is waiting on the answer.
-	client := llmClientFor(lp, 5*time.Minute)
+	// Background: nobody is waiting on an observation, and it must not sit in
+	// front of a voice turn on the single llama-server slot.
+	client := llmBackgroundClientFor(lp, 5*time.Minute)
 	ev := memeval.NewEvaluator(st, st, client, memeval.Config{
 		MaxItems:      cfg.MemoryEval.MaxItems,
 		MinConfidence: cfg.MemoryEval.MinConfidence,

cmd/mavend/modelswap.go

@@ -108,6 +108,34 @@ func wireModelSwap(srv *ipc.Server, phr phraser.Phraser, cfg *config.Config) {
 // rebuilt, so nothing that holds it has to know a swap happened.
 func llmClientFor(lp *phraser.LLMPhraser, timeout time.Duration) *llm.Client {
 	c := llm.New(lp.BaseURL(), timeout)
+	c.SetGate(residentGate, false)
+	lp.OnSwap(func(base string) { c.SetBaseURL(base) })
+	return c
+}
+
+// backgroundQuiet — how long background work stays off the resident model after
+// a foreground request. Long enough to cover the gap between the router call and
+// the phraser call of one turn (router p50 is ~2.7s on this box), short enough
+// that a quiet mailbox is still read promptly.
+const backgroundQuiet = 10 * time.Second
+
+// residentGate — the priority gate on the one llama-server slot, shared by every
+// client llmClientFor builds. Package level because the daemon owns exactly one
+// llama-server: two gates would be two opinions about one queue.
+//
+// The problem it solves: llama-server runs a single slot, so requests queue. Mail
+// extraction is allowed two minutes, and a first poll can hand core 25 messages
+// back to back. Without a gate a voice turn arriving mid-extraction waits for
+// whatever is left of that budget, the router times out into the classifier
+// cascade at its 36.8% floor, and the phraser just waits.
+var residentGate = llm.NewGate(backgroundQuiet)
+
+// llmBackgroundClientFor is llmClientFor for work nobody is waiting on: mail
+// extraction and memory evaluation. Same swap-following client, but it yields
+// to voice turns and only one such request runs at a time.
+func llmBackgroundClientFor(lp *phraser.LLMPhraser, timeout time.Duration) *llm.Client {
+	c := llm.New(lp.BaseURL(), timeout)
+	c.SetGate(residentGate, true)
 	lp.OnSwap(func(base string) { c.SetBaseURL(base) })
 	return c
 }

internal/email/extract.go

@@ -35,6 +35,14 @@ import (
 // search input" — mail is the same class), and Evidence keeps only the subject
 // line, so the review page shows him where a candidate came from without the
 // store growing a copy of his mailbox.
+//
+// One constraint for whoever adds task context to a prompt later: a candidate's
+// text is a model paraphrase of the content of his mail, and it lives in
+// tasks.text. "Maven never sends his mail anywhere" holds today because nothing
+// assembles a context block out of live tasks. The moment something does, mail
+// content reaches whatever that block is sent to, and an outbound search would
+// be sending his mailbox out a paraphrase at a time. Tasks sourced "email:" have
+// to be excluded there, not here.
 
 // MaxCandidates — at most this many candidates per message, enforced by the
 // grammar. A mail with four tasks in it is a mail he has to read himself; a
@@ -81,6 +89,12 @@ func NewExtractor(c Completer, max int, contextBlock func() string) *Extractor {
 	return &Extractor{llm: c, max: max, contextBlock: contextBlock}
 }
 
+// Max — the normalised candidate bound. Exported so the daemon logs what it will
+// actually allow rather than what the config file said: 0 in the config means
+// MaxCandidates here, and logging the raw value said "max 0" and then wrote
+// three.
+func (e *Extractor) Max() int { return e.max }
+
 // extractGrammar — GBNF pinning the answer to a bounded array of fixed-shape
 // candidates. Same reasoning as memeval's evalGrammar and the router's
 // routeGrammar: the shape and the length bound are what keep a small model from
@@ -190,9 +204,13 @@ func renderForModel(msg Message) string {
 	return b.String()
 }
 
-// parseCandidates decodes the grammar-constrained reply, tolerating the
-// wrappers a Thinking model sometimes leaves around it (a fenced block, or
-// leading reasoning before the array).
+// parseCandidates decodes the reply and trims a fenced block or stray prose
+// around the array.
+//
+// Through Extract that tolerance is unreachable: extractGrammar pins the first
+// token to "[", so the model cannot emit reasoning before it. It is kept for
+// callers that pass a raw reply from an ungrammared path, and the note is here
+// so the next reader does not conclude that thinking output is expected.
 func parseCandidates(raw string) ([]Candidate, error) {
 	s := strings.TrimSpace(raw)
 	if i := strings.Index(s, "["); i > 0 {

internal/ipc/api.go

@@ -171,6 +171,12 @@ type IngestMailReq struct {
 // mailbox dedupes to Created=0). Skipped is set when nothing was asked of the
 // model at all — junk, or an empty message.
 //
+// Created == 0 && !Skipped therefore means the model WAS consulted and found no
+// task, which is the common answer. A reader deciding whether to mark a UID
+// seen should treat that the same as a success: asking again would spend the
+// resident model on the same negative answer. Skipped means the same for a
+// different reason. Only an error means "not read yet".
+//
 // Nothing here echoes the mail back. The reader logs counts.
 type IngestMailResp struct {
 	TaskIDs []int64 `json:"task_ids,omitempty"`

internal/llm/client.go

@@ -23,6 +23,32 @@ type Client struct {
 	mu   sync.RWMutex
 	base string
 	http *http.Client
+
+	// gate / background — priority on the single llama-server slot. Set once
+	// at wiring time (SetGate), read on every request. nil gate ⇒ no gating,
+	// which is what every test and every non-daemon caller gets.
+	gate       *Gate
+	background bool
+}
+
+// SetGate gives this client a priority on the shared llama-server slot. Call it
+// immediately after New, before the client is handed to anything: the fields are
+// read under the same lock as base, but the intent is one-time wiring, not a
+// knob to turn at runtime.
+//
+// background = false means "he is waiting for this" and never blocks.
+// background = true means the request yields to voice turns and runs one at a
+// time. See Gate.
+func (c *Client) SetGate(g *Gate, background bool) {
+	c.mu.Lock()
+	c.gate, c.background = g, background
+	c.mu.Unlock()
+}
+
+func (c *Client) gateFor() (*Gate, bool) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.gate, c.background
 }
 
 func New(baseURL string, timeout time.Duration) *Client {
@@ -81,6 +107,17 @@ type resp struct {
 }
 
 func (c *Client) Complete(ctx context.Context, r Req) (string, error) {
+	if g, background := c.gateFor(); g != nil {
+		if background {
+			release, err := g.AcquireBackground(ctx)
+			if err != nil {
+				return "", err
+			}
+			defer release()
+		} else {
+			defer g.Foreground()()
+		}
+	}
 	b, _ := json.Marshal(body{
 		Messages:      []msg{{Role: "system", Content: r.System}, {Role: "user", Content: r.User}},
 		MaxTokens:     r.MaxTokens,

internal/llm/gate.go

@@ -0,0 +1,121 @@
+package llm
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+// Gate — priority access to the one llama-server slot.
+//
+// llama-server is started without -np, so it serves one request at a time and
+// everything else queues. That is fine while every caller is a voice turn, and
+// it stops being fine the moment a background job joins: mail extraction reads
+// up to 4000 characters on a Thinking 1.7B with a two minute budget, and a turn
+// that arrives during one waits for however much of that budget is left. The
+// router degrades to the classifier cascade on error, so he would get the 36.8%
+// floor while his mail is being read, and the phraser has no floor at all and
+// simply waits.
+//
+// So background work asks the gate first:
+//
+//   - at most ONE background request is in flight, whatever the queue depth
+//     upstream. A first poll of a mailbox with 40 unseen messages cannot
+//     serialise 40 extractions ahead of anything.
+//   - a background request waits while any foreground request is in flight, and
+//     for Quiet after the last one finished. The quiet window is what stops an
+//     extraction starting in the gap between the router call and the phraser
+//     call of the same turn.
+//
+// Foreground requests never wait. This is not a fair queue and must not become
+// one: the point is that the thing he is waiting for wins every time.
+//
+// It bounds only what goes through an *llm.Client built with SetGate. The
+// phraser's own HTTP path is not gated, and a turn that reaches the phraser
+// without touching the router is not marked. Every real turn routes first, so
+// the marking is good enough to keep extraction out of the way; it is a
+// courtesy gate, not a scheduler.
+type Gate struct {
+	mu sync.Mutex
+	// fg — foreground requests in flight.
+	fg int
+	// last — when a foreground request last started or finished.
+	last time.Time
+	// bg — one token, so only one background request runs at a time.
+	bg chan struct{}
+
+	quiet time.Duration
+	poll  time.Duration
+	now   func() time.Time
+}
+
+// NewGate returns a gate that holds background work back for quiet after the
+// last foreground request. quiet <= 0 means "wait only while one is in flight".
+func NewGate(quiet time.Duration) *Gate {
+	return &Gate{
+		bg:    make(chan struct{}, 1),
+		quiet: quiet,
+		poll:  50 * time.Millisecond,
+		now:   time.Now,
+	}
+}
+
+// Foreground marks a request as the thing he is waiting for. It never blocks.
+// The returned function must be called when the request finishes.
+func (g *Gate) Foreground() func() {
+	if g == nil {
+		return func() {}
+	}
+	g.mu.Lock()
+	g.fg++
+	g.last = g.now()
+	g.mu.Unlock()
+	return func() {
+		g.mu.Lock()
+		g.fg--
+		g.last = g.now()
+		g.mu.Unlock()
+	}
+}
+
+// AcquireBackground blocks until the slot is free enough for background work,
+// or ctx is done. The returned release function must be called when the request
+// finishes; it is nil on error.
+func (g *Gate) AcquireBackground(ctx context.Context) (func(), error) {
+	if g == nil {
+		return func() {}, nil
+	}
+	select {
+	case g.bg <- struct{}{}:
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+	release := func() { <-g.bg }
+	for {
+		if g.clear() {
+			return release, nil
+		}
+		t := time.NewTimer(g.poll)
+		select {
+		case <-t.C:
+		case <-ctx.Done():
+			t.Stop()
+			release()
+			return nil, ctx.Err()
+		}
+	}
+}
+
+// clear reports whether no foreground request is in flight and the quiet window
+// since the last one has passed.
+func (g *Gate) clear() bool {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	if g.fg > 0 {
+		return false
+	}
+	if g.quiet <= 0 || g.last.IsZero() {
+		return true
+	}
+	return g.now().Sub(g.last) >= g.quiet
+}

internal/llm/gate_test.go

@@ -0,0 +1,101 @@
+package llm
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// Background work must not start while he is waiting on a turn. llama-server
+// serves one request at a time, so an extraction that starts first holds the
+// slot for its whole budget.
+func TestGateBackgroundWaitsForForeground(t *testing.T) {
+	g := NewGate(0)
+	g.poll = time.Millisecond
+	done := g.Foreground()
+
+	started := make(chan struct{})
+	go func() {
+		release, err := g.AcquireBackground(context.Background())
+		if err != nil {
+			t.Errorf("acquire: %v", err)
+			return
+		}
+		close(started)
+		release()
+	}()
+
+	select {
+	case <-started:
+		t.Fatal("background work started while a foreground request was in flight")
+	case <-time.After(20 * time.Millisecond):
+	}
+	done()
+	select {
+	case <-started:
+	case <-time.After(time.Second):
+		t.Fatal("background work never started after the foreground request finished")
+	}
+}
+
+// Only one background request at a time, whatever the queue depth upstream. A
+// first poll of a mailbox with 40 unseen messages must not put 40 extractions
+// on the slot.
+func TestGateOneBackgroundAtATime(t *testing.T) {
+	g := NewGate(0)
+	g.poll = time.Millisecond
+	first, err := g.AcquireBackground(context.Background())
+	if err != nil {
+		t.Fatalf("first: %v", err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+	defer cancel()
+	if _, err := g.AcquireBackground(ctx); err == nil {
+		t.Fatal("a second background request ran alongside the first")
+	}
+	first()
+	second, err := g.AcquireBackground(context.Background())
+	if err != nil {
+		t.Fatalf("second after release: %v", err)
+	}
+	second()
+}
+
+// The quiet window covers the gap between the router call and the phraser call
+// of one turn, so an extraction cannot slip in mid-turn.
+func TestGateQuietWindow(t *testing.T) {
+	now := time.Now()
+	g := NewGate(time.Minute)
+	g.poll = time.Millisecond
+	g.now = func() time.Time { return now }
+	g.Foreground()()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+	defer cancel()
+	if _, err := g.AcquireBackground(ctx); err == nil {
+		t.Fatal("background work started inside the quiet window")
+	}
+	now = now.Add(2 * time.Minute)
+	release, err := g.AcquireBackground(context.Background())
+	if err != nil {
+		t.Fatalf("acquire after the quiet window: %v", err)
+	}
+	release()
+}
+
+// Foreground never waits, whatever else is in flight.
+func TestGateForegroundNeverBlocks(t *testing.T) {
+	g := NewGate(time.Minute)
+	release, err := g.AcquireBackground(context.Background())
+	if err != nil {
+		t.Fatalf("acquire: %v", err)
+	}
+	defer release()
+	done := make(chan struct{})
+	go func() { g.Foreground()(); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("a foreground request waited behind background work")
+	}
+}