diff --git a/cmd/mavend/mail.go b/cmd/mavend/mail.go index 5167e71..dcc28f5 100644 --- a/cmd/mavend/mail.go +++ b/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 diff --git a/cmd/mavend/mail_test.go b/cmd/mavend/mail_test.go index 05a5112..383d64b 100644 --- a/cmd/mavend/mail_test.go +++ b/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)) + } +} diff --git a/cmd/mavend/memoryeval.go b/cmd/mavend/memoryeval.go index 2faea11..f9ad5b2 100644 --- a/cmd/mavend/memoryeval.go +++ b/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, diff --git a/cmd/mavend/modelswap.go b/cmd/mavend/modelswap.go index fae8cb7..f75e29b 100644 --- a/cmd/mavend/modelswap.go +++ b/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 } diff --git a/internal/email/extract.go b/internal/email/extract.go index d96723a..b68d6a8 100644 --- a/internal/email/extract.go +++ b/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 { diff --git a/internal/ipc/api.go b/internal/ipc/api.go index 30cc4ab..38f85ca 100644 --- a/internal/ipc/api.go +++ b/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"` diff --git a/internal/llm/client.go b/internal/llm/client.go index a43748f..e704e36 100644 --- a/internal/llm/client.go +++ b/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, diff --git a/internal/llm/gate.go b/internal/llm/gate.go new file mode 100644 index 0000000..58a3d75 --- /dev/null +++ b/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 +} diff --git a/internal/llm/gate_test.go b/internal/llm/gate_test.go new file mode 100644 index 0000000..7275108 --- /dev/null +++ b/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") + } +}