// mavend/mail.go — core's half of the email reader (Vikunja #246, // docs/plans/01-email-reader.md). // // The split: cmd/mavmaild holds the IMAP credential, connects to the mailbox // and converts messages to plaintext; it hands each message to core over // ipc.MethodIngestMail. Core runs the extraction on the resident model — // llama-server lives in this process, spawned by the phraser — and writes what // comes back through the one task intake seam. // // What this file may produce is exactly one thing: rows in `tasks` with status // "candidate". No fact, no reminder, no note, no nudge, no calendar event. A // 1.7B misreading a mail can therefore put a wrong line on a review page and // nothing else; it can never make Maven speak, and it can never make her // recite something out of an advert as true. // // Off unless configured twice over: no `email` block in mavend.json ⇒ the IPC // method does not exist; no llama-server phraser ⇒ same. A reader pointed at a // core that is not set up for mail gets ErrUnknownMethod rather than silence. package main import ( "context" "fmt" "log" "strings" "time" "unicode" "github.com/kami/maven/internal/config" "github.com/kami/maven/internal/email" "github.com/kami/maven/internal/event" "github.com/kami/maven/internal/ipc" "github.com/kami/maven/internal/phraser" "github.com/kami/maven/internal/store" ) // evidenceMaxChars — how much of the subject line is kept as a candidate's // evidence. Enough to recognise the mail on /tasks, not enough to turn the task // 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 ex *email.Extractor timeout time.Duration now func() time.Time // bus — the unified intake journal (Vikunja #283). This path captures // through the store directly rather than through ipc.CoreAPI, so the // decorator in intake.go does not see it and the publish is explicit here. // nil is a working no-op. bus *event.Bus } // newMailIntake returns nil when mail ingestion must not be available, which is // the default. Both preconditions are real: // // - no cfg.Email ⇒ not configured, and a capability is off unless configured; // - no llama-server phraser ⇒ nothing to extract with. There is deliberately // no keyword fallback: "the subject line became a task" is not extraction, // it is a mailbox rendered as a to-do list, and it would fill the review // page faster than he could clear it. func newMailIntake(st *store.Store, phr phraser.Phraser, cfg *config.Config, bus *event.Bus) *mailIntake { if cfg.Email == nil { return nil } lp, ok := phr.(*phraser.LLMPhraser) if !ok { // 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 } // 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} } // ingest handles one ipc.MethodIngestMail call. // // Junk and empty messages are answered Skipped without touching the model — the // reader's header filter is what keeps the resident model off newsletters. // // Every candidate is captured with Status "candidate", Source "email:" // and the subject as Evidence, under an ExternalID naming the message and the // span it was extracted from. That key is unique over every row whatever its // status, so a mailbox re-read after a restart produces Created=0 — and, more // to the point, a task he already marked done is not re-proposed the next time // the same unread message is read again. 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, Subject: req.Subject, Date: req.Date, Body: req.Body, Junk: req.Junk, } if msg.Junk || (msg.Subject == "" && msg.Body == "") { return ipc.IngestMailResp{Skipped: true}, nil } // 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. return ipc.IngestMailResp{}, fmt.Errorf("mail intake: uid %d: %w", req.UID, err) } if len(cands) == 0 { return ipc.IngestMailResp{}, nil } // 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 for _, c := range cands { t := store.Task{ CreatedTs: now, Text: c.Text, Source: source, Evidence: evidence, // The one status this path may ever write. Anything Maven derived from // something she read is a suggestion until he confirms it on /tasks. Status: store.TaskCandidate, } t.ExternalID = mailExternalID(source, req.UID, c.Text) if due, ok := email.ParseDue(c.Due); ok { t.Due = &due } res, err := m.st.CaptureTask(ctx, t) if err != nil { return resp, fmt.Errorf("mail intake: capture: %w", err) } resp.TaskIDs = append(resp.TaskIDs, res.ID) if res.Created { resp.Created++ // Only a row that was actually created. CaptureTask dedupes on // normalised text among live rows, so a mailbox re-read after a // restart must not refill the journal with tasks already in it. m.bus.Publish(publishableTask(t, now), now) } } // Counts only: the log line names the mailbox and the UID, never the subject, // the sender or the task text. Reviewing a candidate is what /tasks is for. log.Printf("mail intake: %s uid %d → %d candidate(s), %d new", source, req.UID, len(cands), resp.Created) return resp, nil } // wireMailIntake installs the IPC hook, or leaves it nil so the method reports // ErrUnknownMethod. Called on both startup paths (unlocked boot and passkey // unlock) so mail behaves the same either way. func wireMailIntake(srv *ipc.Server, st *store.Store, phr phraser.Phraser, cfg *config.Config, bus *event.Bus) { mi := newMailIntake(st, phr, cfg, bus) if mi == nil { return } srv.IngestMailFn = mi.ingest } // truncateRunes cuts a string to n runes, marking the cut. func truncateRunes(s string, n int) string { r := []rune(s) if len(r) <= n { return s } return string(r[:n]) + "…" } // mailExternalID names the message and the span a candidate was extracted // from. The mailbox and UID identify the message; the normalised text // identifies which of the candidates in it this is, so a message yielding two // tasks gets two keys and a re-read of it gets neither twice. // // UIDs are stable per mailbox, and a mailbox that renumbers (UIDVALIDITY // changing) re-proposes its tasks once, which is the safe direction. func mailExternalID(source string, uid uint32, text string) string { return fmt.Sprintf("%s#%d:%s", source, uid, store.NormalizeTaskText(text)) }