// mavmaild — the mail reader module (Vikunja #246, // docs/plans/01-email-reader.md). // // Every so often it opens one IMAP mailbox read-only, fetches the messages it // has not read yet, and hands each one to core over ipc.MethodIngestMail. Core // runs the extraction on the resident model and writes what comes back as task // CANDIDATES he reviews on /tasks. Nothing here writes to the store, nothing // here can create a reminder, and nothing here speaks. // // Why a separate daemon rather than a loop inside mavend, when extraction has // to happen in mavend anyway: the credential. mavpoll set the precedent with the // zenmoney token (#125) — the module that talks to a third party holds the // secret, reads it from a FILE so it never appears in `ps`, in // docker-compose.yml or in shell history, and core never sees it. Core learns // that mail exists only as message text on one IPC method; it cannot connect to // the mailbox even if it wanted to, and a compromised core yields no mail // password. // // Off unless configured: without -password-file there is nothing to run, and // the daemon says so and exits. If core has no `email` block the very first // ingest comes back ErrUnknownMethod and this daemon stops polling instead of // hammering a socket that will keep refusing. // // Mail is personal, so the log is counts and UIDs: how many messages were // fetched, how many were bulk, how many candidates came back. No subject, no // sender, no body, ever — reviewing a candidate is what /tasks is for. package main import ( "context" "encoding/json" "errors" "flag" "fmt" "log" "os" "os/signal" "path/filepath" "sort" "strings" "syscall" "time" "github.com/kami/maven/internal/email" "github.com/kami/maven/internal/ipc" ) func main() { if err := run(os.Args[1:]); err != nil { fmt.Fprintln(os.Stderr, "mavmaild:", err) os.Exit(1) } } func run(args []string) error { fs := flag.NewFlagSet("mavmaild", flag.ContinueOnError) socket := fs.String("socket", "", "core IPC socket path (required)") server := fs.String("imap", "", "IMAP server, host or host:993 (required)") user := fs.String("user", "", "IMAP username (required)") passFile := fs.String("password-file", "", "file holding the IMAP password (required — never passed as a flag value)") mailbox := fs.String("mailbox", "INBOX", "mailbox to read, read-only") interval := fs.Duration("interval", 15*time.Minute, "how often to read the mailbox") lookback := fs.Duration("lookback", 72*time.Hour, "how far back to search on each poll") // -max and -interval are one decision, not two. Every non-bulk message in a // poll is one serialized llama-server call on core's side, and core gates // mail extraction behind voice turns (llm.Gate), so a large batch does not // mute Maven, it just takes a while. Raise -max only alongside whatever // bound core is running. max := fs.Int("max", 25, "most messages to fetch in one poll") timeout := fs.Duration("timeout", 30*time.Second, "IMAP network timeout") statePath := fs.String("state", "", "file remembering which UIDs were read (default: none — every poll re-reads the window)") if err := fs.Parse(args); err != nil { return err } if *socket == "" { return fmt.Errorf("-socket is required") } if *server == "" || *user == "" || *passFile == "" { return fmt.Errorf("mail reading is off unless configured: set -imap, -user and -password-file") } // The password is read from a file, never taken as a flag value: an argv // secret is visible in `ps` to every user on the box and lands in the compose // file and the shell history. Read once at start — a rotated password means a // restart, which is cheaper than re-reading his credential every quarter hour. raw, err := os.ReadFile(*passFile) if err != nil { return fmt.Errorf("read password file: %w", err) } password := strings.TrimSpace(string(raw)) if password == "" { return fmt.Errorf("password file %s is empty", *passFile) } ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() core, err := ipc.DialWait(*socket, 60*time.Second) if err != nil { return err } defer core.Close() r := &reader{ core: core, addr: *server, user: *user, mailbox: *mailbox, lookback: *lookback, max: *max, timeout: *timeout, state: newSeenState(*statePath), } if err := r.state.load(); err != nil { // A missing or corrupt state file must not stop mail from being read: the // worst case is re-reading the window, and capture dedupes on text. log.Printf("mavmaild: state: %v (starting from an empty seen-set)", err) } // The password is never logged, not even its length. log.Printf("mavmaild: reading %s on %s every %s (lookback %s, max %d/poll)", *mailbox, *server, *interval, *lookback, *max) r.pollOnce(ctx, password) // don't idle a full interval on start t := time.NewTicker(*interval) defer t.Stop() for { select { case <-ctx.Done(): log.Printf("mavmaild: bye") return nil case <-t.C: // Core told us mail ingestion is not configured. Nothing will change // without a core restart, and a restart restarts us too, so the // daemon stays up and does nothing at all. // // It does NOT exit. The compose service inherits restart: // unless-stopped, which restarts a clean exit as readily as a crash, // so exiting here produced a loop: log in to IMAP, get refused by // core, exit, restart, log in again. Four IMAP logins an hour // against a mailbox that has nothing to give, and Gmail and Yandex // both rate-limit exactly that. if r.disabled { continue } r.pollOnce(ctx, password) } } } // mailIngester — the slice of core this daemon uses. One method: hand over a // message. It cannot write a fact, create a reminder or read the store, and the // interface says so. type mailIngester interface { IngestMail(ctx context.Context, req ipc.IngestMailReq) (ipc.IngestMailResp, error) } type reader struct { core mailIngester addr string user string mailbox string lookback time.Duration max int timeout time.Duration state *seenState // fetchMail — the read seam, nil ⇒ the real IMAP read. The tests replace // the whole read rather than the transport: internal/email keeps its dialer // unexported so that no code outside that package can point the reader at a // cleartext socket and hand it the password, and this daemon is code // outside that package. fetchMail func(password string) ([]email.Message, error) // disabled — core answered ErrUnknownMethod, i.e. it has no email block. // Written in pollOnce and read in the ticker loop, both on the one // goroutine that run() drives, so it needs no atomic. If a second caller of // pollOnce ever appears, this becomes a race and has to change. disabled bool } // pollOnce — one read of the mailbox, then one ingest per message. // // A fetch error aborts this poll and nothing else; the next tick tries again. // An ingest error for one message does not skip the rest — one mail the model // choked on should not hide the four behind it. func (r *reader) pollOnce(ctx context.Context, password string) { msgs, err := r.fetch(password) if err != nil { // The error may name a UID; it never names a subject or a sender. log.Printf("mavmaild: fetch: %v", err) if len(msgs) == 0 { return } } var junk, candidates, created int for _, m := range msgs { if ctx.Err() != nil { return } req := ipc.IngestMailReq{ Mailbox: r.mailbox, UID: m.UID, From: m.From, Subject: m.Subject, Date: m.Date, Body: m.Body, } if m.Junk { junk++ // Core is TOLD, which is what its wire doc says: it counts the bulk // message and answers Skipped without spending the model. The header // filter already decided, so no content is sent with the verdict — // nothing will read it. req = ipc.IngestMailReq{Mailbox: r.mailbox, UID: m.UID, Junk: true} } resp, err := r.core.IngestMail(ctx, req) if errors.Is(err, ipc.ErrUnknownMethod) { log.Printf("mavmaild: core has no email block configured — mail ingestion is off; idling until a restart") r.disabled = true return } if err != nil { // Not marked seen: an ingest that failed should be retried next poll. log.Printf("mavmaild: ingest uid %d: %v", m.UID, err) continue } r.state.mark(m.UID) candidates += len(resp.TaskIDs) created += resp.Created } if err := r.state.save(); err != nil { log.Printf("mavmaild: state: %v", err) } log.Printf("mavmaild: %s: %d read, %d bulk, %d candidate(s), %d new", r.mailbox, len(msgs), junk, candidates, created) } // fetch reads the mailbox. Messages already in the seen-set are not fetched at // all, so a steady mailbox costs one SEARCH per poll and nothing else. func (r *reader) fetch(password string) ([]email.Message, error) { if r.fetchMail != nil { return r.fetchMail(password) } f := email.FetchSince{ Addr: r.addr, User: r.user, Mailbox: r.mailbox, Timeout: r.timeout, Since: time.Now().Add(-r.lookback), Max: r.max, Skip: r.state.seen, // Everything below the oldest searchable UID has aged out of the // lookback window and can never be read again. Retiring it is what keeps // one permanently failing message from pinning the high-water mark // forever. See seenState.retire. OnSearch: func(uids []uint32) { if len(uids) == 0 { return } low := uids[0] for _, u := range uids { if u < low { low = u } } r.state.retire(low) }, } return f.Run(password) } // ---- seen state ------------------------------------------------------------ // seenState — the UIDs already handed to core, persisted so a restart does not // re-read (and re-extract, at multi-second LLM cost) the whole lookback window. // // Correctness does not depend on it: ipc.CaptureTask dedupes on normalised text // among live tasks, so a re-read produces no duplicate rows. This exists to save // the model's time, which is why a broken state file is a log line rather than a // failure. // // UIDs are per-mailbox and monotonic, so the set is kept as a high-water mark // plus the stragglers above it. If the server ever changes UIDVALIDITY, UIDs // reset and the window is simply re-read once — dedupe absorbs it. // // The high-water mark only advances through a CONTIGUOUS run, so a UID that // never ingests successfully would pin it forever: everything above stays in // the explicit set, and save rewrites all of it every poll. A year of that is // a few hundred thousand entries written every quarter hour, which breaks // nothing loudly and is exactly why it is worth catching. retire is the answer: // a UID that has fallen out of the SEARCH SINCE window can never be fetched // again, so there is nothing left to wait for. type seenState struct { path string high uint32 set map[uint32]bool dirty bool } func newSeenState(path string) *seenState { return &seenState{path: path, set: map[uint32]bool{}} } type seenFile struct { High uint32 `json:"high"` UIDs []uint32 `json:"uids,omitempty"` } func (s *seenState) seen(uid uint32) bool { return uid <= s.high || s.set[uid] } func (s *seenState) mark(uid uint32) { if s.seen(uid) { return } s.set[uid] = true s.dirty = true // Advance the high-water mark through any contiguous run, so the explicit set // stays small on a mailbox read in order. for { next := s.high + 1 if !s.set[next] { break } delete(s.set, next) s.high = next } } // retire records that no UID below floor is reachable any more — they have // aged out of the lookback window, so no poll will ever fetch them. The // high-water mark can jump past the gap they were holding open, and the // stragglers below it leave the explicit set. // // It never moves backwards, so a UIDVALIDITY reset (UIDs restarting low) makes // this a no-op rather than a way to un-see a mailbox. func (s *seenState) retire(floor uint32) { if floor == 0 || floor-1 <= s.high { return } s.high = floor - 1 for u := range s.set { if u <= s.high { delete(s.set, u) } } // The run above the new mark may now be contiguous with it. for s.set[s.high+1] { delete(s.set, s.high+1) s.high++ } s.dirty = true } func (s *seenState) load() error { if s.path == "" { return nil } b, err := os.ReadFile(s.path) if errors.Is(err, os.ErrNotExist) { return nil // first run } if err != nil { return err } var f seenFile if err := json.Unmarshal(b, &f); err != nil { return fmt.Errorf("parse %s: %w", s.path, err) } s.high = f.High for _, u := range f.UIDs { s.set[u] = true } return nil } // save writes the state atomically (temp file + rename), 0600: it is a list of // message ids from his mailbox, which is metadata about his mail. func (s *seenState) save() error { if s.path == "" || !s.dirty { return nil } uids := make([]uint32, 0, len(s.set)) for u := range s.set { uids = append(uids, u) } sort.Slice(uids, func(i, j int) bool { return uids[i] < uids[j] }) b, err := json.Marshal(seenFile{High: s.high, UIDs: uids}) if err != nil { return err } tmp := s.path + ".tmp" if err := os.MkdirAll(filepath.Dir(s.path), 0o700); err != nil { return err } if err := os.WriteFile(tmp, b, 0o600); err != nil { return err } if err := os.Rename(tmp, s.path); err != nil { return err } s.dirty = false return nil }