From d69a1f80762112717dac6b175e6343b4e269320f Mon Sep 17 00:00:00 2001 From: kami Date: Sat, 1 Aug 2026 14:01:04 +0400 Subject: [PATCH 1/4] email: bound the IMAP read and keep one bad message from blocking the poll The literal size came off the wire with no cap, so the server chose the allocation. A {2147483647} literal was a 2GB make before a byte arrived, and one ordinary mail with a 60MB attachment was 60MB of peak RSS on a box already holding a 1.7B model resident, all of it discarded afterwards by plaintextBody. Literals are now capped at MaxMessageBytes, and a larger one is drained and reported as ErrMessageTooLarge without being kept. Reads are chunked with a deadline refresh, so the timeout is an idle timeout again rather than a budget for the whole message. FetchSince returned on the first fetch error, though its comment described a continue. One oversized message at the top of the window hid every older message behind it, on that poll and on every poll after it. Failures are now collected and the rest of the mailbox is read. An oversized UID is retired as bulk, since it will be the same size next time and the poller marks bulk seen. Timeout zero was accepted and disabled the dial timeout and every socket deadline, which parks the poller forever on a dead server with his credential live in a TLS state. It is now rejected like an empty address. A FETCH answered without a literal was indistinguishable from a vanished message and dropped with no log line. Login now rejects a credential containing a line break instead of stripping it and failing on the server's generic NO. untagged matches the whole key, not a prefix. RunWith is gone: the dial seam is an unexported field again, reachable only through export_test.go, so no code outside the package can hand the reader a cleartext transport and the password. Found in review of #63. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01TrVSBKe3RFDF4fGYKWYQnX --- internal/email/export_test.go | 14 ++++ internal/email/fetch.go | 60 ++++++++++++---- internal/email/fetch_test.go | 78 +++++++++++++++++++-- internal/email/imap.go | 125 +++++++++++++++++++++++++++++++--- internal/email/imap_test.go | 77 ++++++++++++++++++++- 5 files changed, 325 insertions(+), 29 deletions(-) create mode 100644 internal/email/export_test.go diff --git a/internal/email/export_test.go b/internal/email/export_test.go new file mode 100644 index 0000000..2fda570 --- /dev/null +++ b/internal/email/export_test.go @@ -0,0 +1,14 @@ +package email + +import "time" + +// WithDial sets the connection seam for a test. It lives in a _test.go file so +// the seam has no linker symbol in the shipped binary: no code outside this +// package can hand FetchSince a dialer, and therefore no code outside this +// package can point the mail reader at a cleartext transport and give it his +// password. The compiler is what enforces that, which is the whole reason the +// field is unexported. +func (f FetchSince) WithDial(d func(addr string, timeout time.Duration) (*Conn, error)) FetchSince { + f.dial = d + return f +} diff --git a/internal/email/fetch.go b/internal/email/fetch.go index c1c67a7..99c3b8b 100644 --- a/internal/email/fetch.go +++ b/internal/email/fetch.go @@ -1,6 +1,7 @@ package email import ( + "errors" "fmt" "time" ) @@ -26,26 +27,46 @@ type FetchSince struct { Since time.Time Max int Skip func(uid uint32) bool + + // dial — the connection seam, unexported on purpose: see dialer(). Tests + // inside this package set it through export_test.go; nothing outside can. + dial func(addr string, timeout time.Duration) (*Conn, error) +} + +// dial is the connection seam. nil means Dial (implicit TLS); the tests set it +// through export_test.go. It is unexported and there is no exported wrapper +// that takes a dialer, so no code outside this package can point the reader at +// a non-TLS transport and hand it the password. That is a property the compiler +// enforces, not a claim about the callers that happen to exist today. +func (f FetchSince) dialer() func(addr string, timeout time.Duration) (*Conn, error) { + if f.dial != nil { + return f.dial + } + return Dial } // Run performs one read. password is passed here, not stored in the struct, so // the configuration of a mailbox and the secret for it are never the same value // sitting in the same place. +// +// One message that cannot be read does not abandon the poll: the rest of the +// mailbox is still worth reading, and returning early meant one oversized or +// unreadable message permanently hid every older message behind it, poll after +// poll. The returned error joins whatever failed, and the messages that did +// come back come back with it. func (f FetchSince) Run(password string) ([]Message, error) { - return f.RunWith(password, nil) -} - -// RunWith is Run with an explicit connection function, which is how the reader -// daemon and the tests substitute an in-process server. nil ⇒ Dial, i.e. -// implicit TLS with certificate verification; there is no configuration path -// that reaches this, so no deployment can end up talking cleartext IMAP. -func (f FetchSince) RunWith(password string, dial func(addr string, timeout time.Duration) (*Conn, error)) ([]Message, error) { if f.Addr == "" || f.User == "" || f.Mailbox == "" { return nil, fmt.Errorf("email: mailbox not configured (addr/user/mailbox)") } - if dial == nil { - dial = Dial + // Timeout is validated like the other three fields. Zero disables every + // deadline in the path — net.Dialer{Timeout: 0} and a Conn that never calls + // SetDeadline — so a dead server parks the poller forever on a socket read, + // with his credential live in a TLS state. That is the exact thing the + // connect-read-drop shape exists to avoid. + if f.Timeout <= 0 { + return nil, fmt.Errorf("email: timeout must be positive") } + dial := f.dialer() c, err := dial(f.Addr, f.Timeout) if err != nil { return nil, err @@ -77,13 +98,22 @@ func (f FetchSince) RunWith(password string, dial func(addr string, timeout time } out := make([]Message, 0, len(wanted)) + var failed []error for _, uid := range wanted { raw, err := c.Fetch(uid) + if errors.Is(err, ErrMessageTooLarge) { + // Too big to read is a permanent verdict, not a failure to retry: + // the message will be the same size next poll. Carried as bulk so + // the poller marks it seen and stops fetching it, exactly like a + // newsletter. Nothing is sent to the model. + out = append(out, Message{UID: uid, Junk: true, JunkReason: "oversize"}) + continue + } if err != nil { - // One unreadable message does not abandon the poll; the rest of the - // mailbox is still worth reading. The error names the UID, not the - // message. - return out, fmt.Errorf("email: fetch uid %d: %w", uid, err) + // The error names the UID, never the message. Collected rather than + // returned, so the messages behind this one are still read. + failed = append(failed, fmt.Errorf("email: fetch uid %d: %w", uid, err)) + continue } if len(raw) == 0 { continue // vanished between SEARCH and FETCH @@ -94,5 +124,5 @@ func (f FetchSince) RunWith(password string, dial func(addr string, timeout time } out = append(out, msg) } - return out, nil + return out, errors.Join(failed...) } diff --git a/internal/email/fetch_test.go b/internal/email/fetch_test.go index fa09fb2..f8cd720 100644 --- a/internal/email/fetch_test.go +++ b/internal/email/fetch_test.go @@ -22,11 +22,7 @@ func TestFetchSinceRun(t *testing.T) { Max: 2, Skip: func(uid uint32) bool { return uid == 3 }, } - msgs, err := fs.RunWith("secret", func(addr string, timeout time.Duration) (*Conn, error) { - cli, srv := net.Pipe() - go f.serve(t, srv) - return NewConn(cli, timeout) - }) + msgs, err := fs.WithDial(dialer(t, f)).Run("secret") if err != nil { t.Fatalf("run: %v", err) } @@ -47,3 +43,75 @@ func TestFetchSinceRequiresConfig(t *testing.T) { t.Fatal("an unconfigured mailbox must not be read") } } + +// Timeout zero disables the dial timeout AND every socket deadline, so a dead +// server parks the poller forever with his credential live in a TLS state. +func TestFetchSinceRejectsZeroTimeout(t *testing.T) { + fs := FetchSince{Addr: "mail.example:993", User: "kami", Mailbox: "INBOX"} + if _, err := fs.Run("secret"); err == nil { + t.Fatal("a zero timeout must be rejected like an empty address") + } +} + +// dialer wires a client Conn to an in-process fake over net.Pipe. +func dialer(t *testing.T, f *fakeIMAP) func(string, time.Duration) (*Conn, error) { + t.Helper() + return func(addr string, timeout time.Duration) (*Conn, error) { + cli, srv := net.Pipe() + go f.serve(t, srv) + return NewConn(cli, timeout) + } +} + +// One message that cannot be read must not hide the older ones behind it. The +// old code returned on the first failure, so an oversized or unreadable UID +// blocked every message below it on every poll, forever. +func TestFetchSinceContinuesPastABadMessage(t *testing.T) { + mk := func(subject string) string { + return "Subject: " + subject + "\r\nContent-Type: text/plain; charset=utf-8\r\n\r\nbody\r\n" + } + f := &fakeIMAP{ + uids: []uint32{1, 2, 3}, + msgs: map[uint32]string{1: mk("one"), 3: mk("three")}, + quoted: map[uint32]bool{2: true}, + } + fs := FetchSince{ + Addr: "mail.example:993", User: "kami", Mailbox: "INBOX", + Timeout: 5 * time.Second, + Since: time.Date(2026, 7, 30, 0, 0, 0, 0, time.UTC), + } + msgs, err := fs.WithDial(dialer(t, f)).Run("secret") + if err == nil { + t.Fatal("the unreadable UID must still be reported") + } + if !strings.Contains(err.Error(), "uid 2") { + t.Errorf("error should name the UID: %v", err) + } + if len(msgs) != 2 { + t.Fatalf("got %d messages, want the two readable ones: %+v", len(msgs), msgs) + } + if msgs[0].Subject != "three" || msgs[1].Subject != "one" { + t.Errorf("subjects = %q,%q, want three,one", msgs[0].Subject, msgs[1].Subject) + } +} + +// An oversized message is retired as bulk rather than retried: it will be the +// same size next poll, and the poller marks bulk seen without a model call. +func TestFetchSinceRetiresOversizedMessage(t *testing.T) { + f := &fakeIMAP{uids: []uint32{7}, oversize: map[uint32]int{7: MaxMessageBytes + 1}} + fs := FetchSince{ + Addr: "mail.example:993", User: "kami", Mailbox: "INBOX", + Timeout: 5 * time.Second, + Since: time.Date(2026, 7, 30, 0, 0, 0, 0, time.UTC), + } + msgs, err := fs.WithDial(dialer(t, f)).Run("secret") + if err != nil { + t.Fatalf("run: %v", err) + } + if len(msgs) != 1 || !msgs[0].Junk || msgs[0].JunkReason != "oversize" { + t.Fatalf("want one oversize-bulk message, got %+v", msgs) + } + if msgs[0].Body != "" || msgs[0].Subject != "" { + t.Error("nothing from an oversized message may be kept") + } +} diff --git a/internal/email/imap.go b/internal/email/imap.go index 44f2150..845eb40 100644 --- a/internal/email/imap.go +++ b/internal/email/imap.go @@ -3,6 +3,7 @@ package email import ( "bufio" "crypto/tls" + "errors" "fmt" "io" "net" @@ -32,6 +33,31 @@ import ( // option to get it wrong once. const DefaultIMAPPort = "993" +// MaxMessageBytes — the largest message this client will read into memory. +// +// The literal size comes off the wire, so an unbounded read is an allocation +// the server picks: "{2147483647}" is a 2GB make() before a single byte +// arrives, and one ordinary mail with a 60MB attachment is a 60MB peak RSS on a +// box already holding a 1.7B model resident. All of it would then be thrown +// away, because plaintextBody skips attachments and the body is truncated to +// MaxBodyBytes anyway. +// +// 2 MiB is well above what prose plus quoted history plus base64 HTML needs and +// well below what hurts. A larger message is drained and reported as +// ErrMessageTooLarge rather than read. +const MaxMessageBytes = 2 << 20 + +// readChunk — how much of a literal is read between deadline refreshes. The +// per-connection timeout must stay an IDLE timeout: with one deadline around +// the whole read it becomes a whole-message budget, and a healthy but slow +// uplink then fails the same message on every poll forever. +const readChunk = 64 << 10 + +// ErrMessageTooLarge — the server announced a literal above MaxMessageBytes. +// The connection stays usable (the bytes are drained), and the caller decides +// what to do with the UID. FetchSince retires it rather than retrying it. +var ErrMessageTooLarge = errors.New("email: message larger than the read cap") + // Conn — one authenticated IMAP connection. Not safe for concurrent use; the // poller drives one connection at a time. type Conn struct { @@ -78,6 +104,18 @@ func (c *Conn) Close() error { return c.rwc.Close() } // never stored on the Conn: nothing in this package keeps a credential alive // past the command that uses it, so no struct dump or panic trace can carry it. func (c *Conn) Login(user, pass string) error { + // A credential with a line break in it is rejected here, not silently + // repaired. quote() strips CR and LF so a stray newline can never become a + // second command, but stripping alone means a password file that picked up + // a newline authenticates as a DIFFERENT string and comes back as the + // server's generic NO, which is a long debugging session. This error names + // the problem and cannot leak the value. + if strings.ContainsAny(user, "\r\n") { + return fmt.Errorf("email: login: username contains a line break") + } + if strings.ContainsAny(pass, "\r\n") { + return fmt.Errorf("email: login: password contains a line break") + } // The command line itself is never logged (see exec) — a LOGIN line IS the // credential. if _, err := c.exec(fmt.Sprintf("LOGIN %s %s", quote(user), quote(pass))); err != nil { @@ -130,34 +168,60 @@ var literalSize = regexp.MustCompile(`\{(\d+)\}$`) // Fetch returns the raw RFC 5322 bytes of one message, by UID. // // Returns (nil, nil) when the UID no longer exists — a message he deleted -// between SEARCH and FETCH is normal, not an error. +// between SEARCH and FETCH is normal, not an error. A FETCH response that came +// back with no literal in it is NOT that case and is an error, so a message +// that exists and was readable is never dropped without a log line. +// +// A literal above MaxMessageBytes is drained without being kept and reported as +// ErrMessageTooLarge. func (c *Conn) Fetch(uid uint32) ([]byte, error) { tag := c.nextTag() if err := c.send(fmt.Sprintf("%s UID FETCH %d (BODY.PEEK[])", tag, uid)); err != nil { return nil, err } var raw []byte + var sawFetch, tooLarge bool for { line, err := c.readLine() if err != nil { return nil, fmt.Errorf("email: fetch %d: %w", uid, err) } if done, err := c.tagged(tag, line); done { - if err != nil { + switch { + case err != nil: return nil, fmt.Errorf("email: fetch %d: %w", uid, err) + case tooLarge: + return nil, fmt.Errorf("email: fetch %d: %w", uid, ErrMessageTooLarge) + case sawFetch && raw == nil: + // The server answered for this UID but not with a literal (a + // quoted string, say). Silently skipping it would look exactly + // like a vanished message. + return nil, fmt.Errorf("email: fetch %d: no message literal in the FETCH response", uid) } return raw, nil } + if strings.HasPrefix(line, "* ") && strings.Contains(line, " FETCH ") { + sawFetch = true + } m := literalSize.FindStringSubmatch(strings.TrimSpace(line)) if m == nil { continue } n, err := strconv.Atoi(m[1]) - if err != nil { + if err != nil || n < 0 { continue } - buf := make([]byte, n) - if _, err := io.ReadFull(c.r, buf); err != nil { + if n > MaxMessageBytes { + // Drained rather than read: the stream has to stay aligned for the + // tagged completion, but nothing is allocated and nothing is parsed. + tooLarge = true + if err := c.discard(int64(n)); err != nil { + return nil, fmt.Errorf("email: fetch %d: drain literal: %w", uid, err) + } + continue + } + buf, err := c.readN(n) + if err != nil { return nil, fmt.Errorf("email: fetch %d: literal: %w", uid, err) } if raw == nil { @@ -203,8 +267,8 @@ func (c *Conn) exec(cmd string) ([]string, error) { // send asks for one outside Fetch, but skip it if it appears so the // stream stays aligned. if m := literalSize.FindStringSubmatch(strings.TrimSpace(line)); m != nil { - if n, err := strconv.Atoi(m[1]); err == nil { - if _, err := io.CopyN(io.Discard, c.r, int64(n)); err != nil { + if n, err := strconv.Atoi(m[1]); err == nil && n > 0 { + if err := c.discard(int64(n)); err != nil { return nil, err } } @@ -238,6 +302,43 @@ func (c *Conn) send(line string) error { return nil } +// readN reads exactly n bytes, refreshing the deadline every readChunk so the +// timeout stays an idle timeout rather than a budget for the whole literal. +func (c *Conn) readN(n int) ([]byte, error) { + buf := make([]byte, n) + for off := 0; off < n; { + end := off + readChunk + if end > n { + end = n + } + c.setDeadline() + got, err := io.ReadFull(c.r, buf[off:end]) + off += got + if err != nil { + return nil, err + } + } + return buf, nil +} + +// discard throws away n bytes of literal, same chunked deadline refresh as +// readN and no allocation proportional to n. +func (c *Conn) discard(n int64) error { + for n > 0 { + chunk := int64(readChunk) + if chunk > n { + chunk = n + } + c.setDeadline() + got, err := io.CopyN(io.Discard, c.r, chunk) + n -= got + if err != nil { + return err + } + } + return nil +} + func (c *Conn) readLine() (string, error) { c.setDeadline() line, err := c.r.ReadString('\n') @@ -259,6 +360,10 @@ func (c *Conn) setDeadline() { } // untagged splits "* SEARCH 1 2 3" into its payload when the key matches. +// +// The key must be the whole word: a prefix test would also match a future +// extension's "* SEARCHRES", and reading its payload as UIDs is the kind of +// thing that ages badly next to an IMAP capability nobody asked for. func untagged(line, key string) (string, bool) { if !strings.HasPrefix(line, "* ") { return "", false @@ -267,7 +372,11 @@ func untagged(line, key string) (string, bool) { if !strings.HasPrefix(rest, key) { return "", false } - return strings.TrimSpace(rest[len(key):]), true + rest = rest[len(key):] + if rest != "" && rest[0] != ' ' && rest[0] != '\t' { + return "", false + } + return strings.TrimSpace(rest), true } // quote renders an IMAP quoted string. Passwords routinely contain characters diff --git a/internal/email/imap_test.go b/internal/email/imap_test.go index cd82d08..55dd412 100644 --- a/internal/email/imap_test.go +++ b/internal/email/imap_test.go @@ -2,6 +2,7 @@ package email import ( "bufio" + "errors" "fmt" "net" "strconv" @@ -18,6 +19,11 @@ type fakeIMAP struct { uids []uint32 cmds []string failOn string // substring of a command to answer NO + // oversize — UIDs answered with a literal of this many bytes, which the + // server then actually sends. Used to exercise the read cap. + oversize map[uint32]int + // quoted — UIDs answered with a quoted string instead of a literal. + quoted map[uint32]bool } func (f *fakeIMAP) serve(t *testing.T, c net.Conn) { @@ -54,7 +60,20 @@ func (f *fakeIMAP) serve(t *testing.T, c net.Conn) { fmt.Fprintf(c, "%s OK search done\r\n", tag) case strings.HasPrefix(upper, "UID FETCH"): uid64, _ := strconv.ParseUint(strings.Fields(cmd)[2], 10, 32) - raw, ok := f.msgs[uint32(uid64)] + uid := uint32(uid64) + if n, big := f.oversize[uid]; big { + fmt.Fprintf(c, "* 1 FETCH (UID %d BODY[] {%d}\r\n", uid64, n) + fmt.Fprint(c, strings.Repeat("x", n)) + fmt.Fprint(c, ")\r\n") + fmt.Fprintf(c, "%s OK fetch done\r\n", tag) + continue + } + if f.quoted[uid] { + fmt.Fprintf(c, "* 1 FETCH (UID %d BODY[] \"short\")\r\n", uid64) + fmt.Fprintf(c, "%s OK fetch done\r\n", tag) + continue + } + raw, ok := f.msgs[uid] if ok { fmt.Fprintf(c, "* 1 FETCH (UID %d BODY[] {%d}\r\n", uid64, len(raw)) fmt.Fprint(c, raw) @@ -160,3 +179,59 @@ func TestQuoteStripsNewlines(t *testing.T) { t.Errorf("quote kept a line break: %q", got) } } + +// A credential with a line break in it is a broken password file, not a +// password. Stripping it silently authenticates as a different string and the +// server answers its generic NO. +func TestLoginRejectsCredentialWithNewline(t *testing.T) { + f := &fakeIMAP{} + c := dialFake(t, f) + err := c.Login("kami", "s3cr3t\nA1 LOGOUT") + if err == nil { + t.Fatal("a password with a line break must be rejected") + } + if strings.Contains(err.Error(), "s3cr3t") { + t.Errorf("error leaks the password: %v", err) + } + if len(f.cmds) != 0 { + t.Errorf("nothing should have been sent, got %v", f.cmds) + } +} + +// The literal size comes off the wire. Without a cap the server picks the +// allocation, and one 60MB attachment is 60MB of peak RSS on a box holding a +// 1.7B model, all of it thrown away by plaintextBody afterwards. +func TestFetchRefusesOversizedLiteral(t *testing.T) { + f := &fakeIMAP{uids: []uint32{1}, oversize: map[uint32]int{1: MaxMessageBytes + 1}} + c := dialFake(t, f) + raw, err := c.Fetch(1) + if !errors.Is(err, ErrMessageTooLarge) { + t.Fatalf("fetch err = %v, want ErrMessageTooLarge", err) + } + if raw != nil { + t.Errorf("an oversized message must not be kept, got %d bytes", len(raw)) + } + // The stream stayed aligned: the connection is still usable. + if err := c.Select("INBOX"); err != nil { + t.Errorf("connection unusable after draining: %v", err) + } +} + +// A FETCH that answered without a literal is not a vanished message, and must +// not be skipped as silently as one. +func TestFetchNonLiteralResponseIsAnError(t *testing.T) { + f := &fakeIMAP{uids: []uint32{1}, quoted: map[uint32]bool{1: true}} + c := dialFake(t, f) + if _, err := c.Fetch(1); err == nil { + t.Fatal("a FETCH response with no literal must be reported, not dropped") + } +} + +func TestUntaggedMatchesWholeKeyOnly(t *testing.T) { + if _, ok := untagged("* SEARCHRES 1 2 3", "SEARCH"); ok { + t.Error("SEARCH must not match SEARCHRES") + } + if rest, ok := untagged("* SEARCH 1 2 3", "SEARCH"); !ok || rest != "1 2 3" { + t.Errorf("untagged = (%q, %v), want (\"1 2 3\", true)", rest, ok) + } +} From 6c81df17ecf930b4ce12c7ac5ac3431d2bd4b1d1 Mon Sep 17 00:00:00 2001 From: kami Date: Sat, 1 Aug 2026 14:01:04 +0400 Subject: [PATCH 2/4] email: drop the dead Gmail rule, fix nested MIME, decode windows-1251 The Gmail category rule matched X-GM-LABELS and X-Gmail-Labels against the parsed header block. Neither is a header. X-GM-LABELS is a Gmail FETCH data item and never appears in the message source, and X-Gmail-Labels only exists in a Takeout export, so the branch could not fire against a real mailbox while its doc comment promised a Promotions filter. Its test built the header by hand and therefore asserted the matcher rather than the plumbing. The rule is removed and the comment says what bringing it back would take. multipartText folded a nested multipart's answer into one string, so HTML derived text landed in the plain bucket and a real text/plain sibling later in the message was discarded by the guard on plain being set. The two buckets now stay separate through the recursion. windows-1251 returned an unsupported-charset error and the message degraded to subject only. That is the charset older Russian senders still use, so those mails could never produce a task candidate. It is decoded from a 128 entry table here rather than by vendoring x/text, for the body and for encoded words in the subject. Every other unknown charset still degrades to subject only. Found in review of #63. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01TrVSBKe3RFDF4fGYKWYQnX --- internal/email/charset.go | 48 +++++++++++++++ internal/email/junk.go | 33 ++++------- internal/email/junk_test.go | 6 +- internal/email/message.go | 93 +++++++++++++++++++++--------- internal/email/message_test.go | 46 ++++++++++++++- internal/email/testdata/cp1251.eml | 4 +- internal/email/testdata/koi8r.eml | 7 +++ 7 files changed, 182 insertions(+), 55 deletions(-) create mode 100644 internal/email/charset.go create mode 100644 internal/email/testdata/koi8r.eml diff --git a/internal/email/charset.go b/internal/email/charset.go new file mode 100644 index 0000000..f70718a --- /dev/null +++ b/internal/email/charset.go @@ -0,0 +1,48 @@ +package email + +// windows-1251 (and its ASCII-compatible low half) is decoded here rather than +// pulled in from x/text. +// +// The alternative was returning an error for the charset, which ParseMessage +// turns into a subject-only message. That is a live gap and not a small one: +// cp1251 is still what older Russian senders emit, and subject-only means those +// mails can never produce a task candidate. The whole of x/text/encoding is a +// large dependency for the most privacy-sensitive path in the tree, and +// windows-1251 is a 128-entry table. +// +// Only cp1251 is added. Guessing at an unknown charset stays forbidden: mojibake +// is worse than nothing, because the model extracts a task from it happily. + +// cp1251High — the 0x80..0xFF half of windows-1251. The low half is ASCII. +var cp1251High = [128]rune{ + 0x0402, 0x0403, 0x201A, 0x0453, 0x201E, 0x2026, 0x2020, 0x2021, + 0x20AC, 0x2030, 0x0409, 0x2039, 0x040A, 0x040C, 0x040B, 0x040F, + 0x0452, 0x2018, 0x2019, 0x201C, 0x201D, 0x2022, 0x2013, 0x2014, + 0xFFFD, 0x2122, 0x0459, 0x203A, 0x045A, 0x045C, 0x045B, 0x045F, + 0x00A0, 0x040E, 0x045E, 0x0408, 0x00A4, 0x0490, 0x00A6, 0x00A7, + 0x0401, 0x00A9, 0x0404, 0x00AB, 0x00AC, 0x00AD, 0x00AE, 0x0407, + 0x00B0, 0x00B1, 0x0406, 0x0456, 0x0491, 0x00B5, 0x00B6, 0x00B7, + 0x0451, 0x2116, 0x0454, 0x00BB, 0x0458, 0x0405, 0x0455, 0x0457, + 0x0410, 0x0411, 0x0412, 0x0413, 0x0414, 0x0415, 0x0416, 0x0417, + 0x0418, 0x0419, 0x041A, 0x041B, 0x041C, 0x041D, 0x041E, 0x041F, + 0x0420, 0x0421, 0x0422, 0x0423, 0x0424, 0x0425, 0x0426, 0x0427, + 0x0428, 0x0429, 0x042A, 0x042B, 0x042C, 0x042D, 0x042E, 0x042F, + 0x0430, 0x0431, 0x0432, 0x0433, 0x0434, 0x0435, 0x0436, 0x0437, + 0x0438, 0x0439, 0x043A, 0x043B, 0x043C, 0x043D, 0x043E, 0x043F, + 0x0440, 0x0441, 0x0442, 0x0443, 0x0444, 0x0445, 0x0446, 0x0447, + 0x0448, 0x0449, 0x044A, 0x044B, 0x044C, 0x044D, 0x044E, 0x044F, +} + +// decodeCP1251 maps each byte through the table. Every byte has a defined +// meaning in this charset, so decoding cannot fail. +func decodeCP1251(b []byte) string { + out := make([]rune, 0, len(b)) + for _, c := range b { + if c < 0x80 { + out = append(out, rune(c)) + continue + } + out = append(out, cp1251High[c-0x80]) + } + return string(out) +} diff --git a/internal/email/junk.go b/internal/email/junk.go index dc64167..4895430 100644 --- a/internal/email/junk.go +++ b/internal/email/junk.go @@ -21,8 +21,18 @@ import ( // - Auto-Submitted other than "no" (RFC 3834) — generated by a machine. // - X-Spam-Flag: YES, X-Spam-Status: Yes — the spam filter upstream already // decided; we do not second-guess it in the other direction. -// - X-GM-LABELS / X-Gmail-Labels containing a Gmail category — Gmail's own -// Promotions/Social/Forums/Spam classification, when the server sends it. +// +// There is deliberately NO Gmail-category rule. One was written and removed: +// it matched X-GM-LABELS and X-Gmail-Labels against the parsed header block, +// and neither is a header. X-GM-LABELS is a Gmail FETCH data item, requested as +// "UID FETCH n (X-GM-LABELS)" and never present in the message source; +// X-Gmail-Labels only exists in a Takeout mbox export. This client asks for +// BODY.PEEK[] and nothing else, so the rule could not fire against a real +// mailbox while its doc comment promised a Promotions filter. Gmail's promotion +// mail carries List-Unsubscribe in practice and is caught by the rule above. +// Bringing the category rule back means adding the FETCH item and carrying the +// labels into classifyJunk out of band, not matching a header that never +// arrives. // // Deliberately NOT here: sender allow/deny lists and subject keyword matching. // Both are configuration that ages badly and both would be a place for his @@ -35,16 +45,6 @@ import ( // junkHeaders — headers whose mere presence marks bulk mail. var junkPresence = []string{"List-Unsubscribe", "List-Id", "List-Post"} -// gmailCategories — Gmail's category labels, lowercased as they appear in -// X-GM-LABELS. "important" and "inbox" are labels too, and are NOT categories. -// Matching is by these exact tokens (substring is fine — they are namespaced -// and cannot appear in a hand-made label by accident), so a user label named -// "Social Club" is not mistaken for Gmail's Social category. -var gmailCategories = []string{ - "category_promotions", "category_social", "category_forums", "category_updates", - `\spam`, `\junk`, -} - // classifyJunk returns whether the message is bulk/automated and why. The // reason is a short header name, safe to log — it names the marker, never the // sender or the subject. @@ -67,14 +67,5 @@ func classifyJunk(h mail.Header) (bool, string) { if v := strings.ToLower(strings.TrimSpace(h.Get("X-Spam-Status"))); strings.HasPrefix(v, "yes") { return true, "x-spam-status" } - labels := strings.ToLower(h.Get("X-GM-LABELS") + " " + h.Get("X-Gmail-Labels")) - for _, c := range gmailCategories { - if c == "" { - continue - } - if strings.Contains(labels, c) { - return true, "gmail-category" - } - } return false, "" } diff --git a/internal/email/junk_test.go b/internal/email/junk_test.go index d8e8722..81645c9 100644 --- a/internal/email/junk_test.go +++ b/internal/email/junk_test.go @@ -31,8 +31,10 @@ func TestClassifyJunk(t *testing.T) { {"spam flag", "From: a@b.c\nX-Spam-Flag: YES", true, "x-spam-flag"}, {"spam status", "From: a@b.c\nX-Spam-Status: Yes, score=9.1", true, "x-spam-status"}, {"spam status no", "From: a@b.c\nX-Spam-Status: No, score=0.1", false, ""}, - {"gmail promo", "From: a@b.c\nX-Gmail-Labels: Inbox,CATEGORY_PROMOTIONS", true, "gmail-category"}, - {"user label", "From: a@b.c\nX-Gmail-Labels: Social Club,Important", false, ""}, + // X-GM-LABELS is a Gmail FETCH data item, not a header, so it never + // reaches classifyJunk through this client. The rule that matched it was + // removed rather than left claiming a Promotions filter that never ran. + {"gmail label header is not a rule", "From: a@b.c\nX-Gmail-Labels: Inbox,CATEGORY_PROMOTIONS", false, ""}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { diff --git a/internal/email/message.go b/internal/email/message.go index 883be39..5219d39 100644 --- a/internal/email/message.go +++ b/internal/email/message.go @@ -94,7 +94,14 @@ func plaintextBody(contentType, encoding string, body io.Reader) (string, error) if boundary == "" { return "", fmt.Errorf("email: multipart without boundary") } - return multipartText(multipart.NewReader(body, boundary)) + plain, html, err := multipartText(multipart.NewReader(body, boundary)) + if err != nil { + return "", err + } + if strings.TrimSpace(plain) != "" { + return plain, nil + } + return html, nil case mediaType == "text/html": raw, err := decodeBody(body, encoding, params["charset"]) if err != nil { @@ -109,10 +116,15 @@ func plaintextBody(contentType, encoding string, body io.Reader) (string, error) } } -// multipartText reads one multipart level, recursing into nested multiparts. -// Returns the plain part if any part yielded one, else the stripped HTML. -func multipartText(mr *multipart.Reader) (string, error) { - var plain, html string +// multipartText reads one multipart level, recursing into nested multiparts, +// and returns the two buckets separately: real text/plain, and text stripped +// out of HTML. +// +// The buckets stay separate all the way up because a nested multipart can +// contribute either kind. Folding a nested level's answer into one string put +// HTML-derived text in the plain bucket, and a real text/plain sibling later in +// the message was then thrown away by the "plain is already set" guard. +func multipartText(mr *multipart.Reader) (plain, html string, err error) { for { part, err := mr.NextPart() if err == io.EOF { @@ -127,35 +139,47 @@ func multipartText(mr *multipart.Reader) (string, error) { continue // attachment } ct := part.Header.Get("Content-Type") - mediaType, _, _ := mime.ParseMediaType(ct) - text, err := plaintextBody(ct, part.Header.Get("Content-Transfer-Encoding"), part) - part.Close() - if err != nil || strings.TrimSpace(text) == "" { - continue - } - if mediaType == "text/html" && !strings.HasPrefix(mediaType, "multipart/") { - if html == "" { - html = text + mediaType, params, _ := mime.ParseMediaType(ct) + switch { + case strings.HasPrefix(mediaType, "multipart/"): + var np, nh string + if b := params["boundary"]; b != "" { + np, nh, _ = multipartText(multipart.NewReader(part, b)) + } + part.Close() + if plain == "" { + plain = np + } + if html == "" { + html = nh + } + default: + text, terr := plaintextBody(ct, part.Header.Get("Content-Transfer-Encoding"), part) + part.Close() + if terr != nil || strings.TrimSpace(text) == "" { + continue + } + if mediaType == "text/html" { + if html == "" { + html = text + } + continue + } + if plain == "" { + plain = text } - continue - } - if plain == "" { - plain = text } } - if strings.TrimSpace(plain) != "" { - return plain, nil - } - return html, nil + return plain, html, nil } // decodeBody applies the transfer encoding, then the charset. // -// Charset support is UTF-8 (and ASCII, its subset) only, on purpose: x/text's -// encoding tables are not vendored here, and guessing at windows-1251 bytes -// would feed the model mojibake it would happily extract a task from. An -// unsupported charset returns an error, which ParseMessage turns into an empty -// body — subject-only, which is honest. +// Charset support is UTF-8 (and ASCII, its subset) plus windows-1251, which is +// decoded from a table in charset.go — see the reasoning there. Everything else +// returns an error, which ParseMessage turns into an empty body: subject-only, +// which is honest. Guessing at unknown bytes would feed the model mojibake it +// would happily extract a task from. func decodeBody(r io.Reader, encoding, charset string) (string, error) { switch strings.ToLower(strings.TrimSpace(encoding)) { case "quoted-printable": @@ -170,6 +194,8 @@ func decodeBody(r io.Reader, encoding, charset string) (string, error) { switch cs := strings.ToLower(strings.TrimSpace(charset)); cs { case "", "utf-8", "utf8", "us-ascii", "ascii": return string(b), nil + case "windows-1251", "cp1251", "windows1251", "x-cp1251": + return decodeCP1251(b), nil default: return "", fmt.Errorf("email: unsupported charset %q", cs) } @@ -181,6 +207,19 @@ func decodeBody(r io.Reader, encoding, charset string) (string, error) { // shown to him as evidence. func decodeHeader(v string) string { dec := new(mime.WordDecoder) + // Same charset support as the body: an old Russian sender encodes the + // subject in windows-1251 too, and a subject is often the whole task. + dec.CharsetReader = func(charset string, r io.Reader) (io.Reader, error) { + switch strings.ToLower(strings.TrimSpace(charset)) { + case "windows-1251", "cp1251", "windows1251", "x-cp1251": + b, err := io.ReadAll(io.LimitReader(r, 1<<16)) + if err != nil && len(b) == 0 { + return nil, err + } + return strings.NewReader(decodeCP1251(b)), nil + } + return nil, fmt.Errorf("email: unsupported charset %q", charset) + } out, err := dec.DecodeHeader(v) if err != nil { return collapse(v) diff --git a/internal/email/message_test.go b/internal/email/message_test.go index 0ef9028..c961dce 100644 --- a/internal/email/message_test.go +++ b/internal/email/message_test.go @@ -70,13 +70,28 @@ func TestParsePrefersPlainAndSkipsAttachments(t *testing.T) { } } -// An unsupported charset must degrade to headers-only rather than to mojibake -// the model would then extract a task from. -func TestParseUnsupportedCharsetKeepsHeaders(t *testing.T) { +// windows-1251 is what older Russian senders still emit. Subject-only for those +// mails meant they could never produce a task candidate. +func TestParseCP1251(t *testing.T) { msg, err := ParseMessage(3, fixture(t, "cp1251.eml")) if err != nil { t.Fatalf("parse: %v", err) } + if want := "Счёт за интернет"; msg.Subject != want { + t.Errorf("subject = %q, want %q", msg.Subject, want) + } + if want := "Оплати счёт до пятницы."; !strings.Contains(msg.Body, want) { + t.Errorf("body = %q, want it to contain %q", msg.Body, want) + } +} + +// A charset with no table here must degrade to headers-only rather than to +// mojibake the model would then extract a task from. +func TestParseUnsupportedCharsetKeepsHeaders(t *testing.T) { + msg, err := ParseMessage(3, fixture(t, "koi8r.eml")) + if err != nil { + t.Fatalf("parse: %v", err) + } if msg.Subject != "Legacy" { t.Errorf("subject = %q, want Legacy", msg.Subject) } @@ -85,6 +100,31 @@ func TestParseUnsupportedCharsetKeepsHeaders(t *testing.T) { } } +// A nested multipart/alternative that only had HTML must not fill the plain +// bucket: a real text/plain sibling later in the message is the better text and +// used to be discarded. +func TestParseNestedHTMLDoesNotShadowLaterPlain(t *testing.T) { + raw := "Subject: nested\r\n" + + "Content-Type: multipart/mixed; boundary=OUT\r\n\r\n" + + "--OUT\r\n" + + "Content-Type: multipart/alternative; boundary=IN\r\n\r\n" + + "--IN\r\n" + + "Content-Type: text/html; charset=utf-8\r\n\r\n" + + "

from the html part

\r\n" + + "--IN--\r\n" + + "--OUT\r\n" + + "Content-Type: text/plain; charset=utf-8\r\n\r\n" + + "the real plain text\r\n" + + "--OUT--\r\n" + msg, err := ParseMessage(5, []byte(raw)) + if err != nil { + t.Fatalf("parse: %v", err) + } + if got := strings.TrimSpace(msg.Body); got != "the real plain text" { + t.Errorf("body = %q, want the text/plain part to win", got) + } +} + func TestParseTruncatesLongBody(t *testing.T) { var b strings.Builder b.WriteString("Subject: long\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n") diff --git a/internal/email/testdata/cp1251.eml b/internal/email/testdata/cp1251.eml index ca1a016..5f987cf 100644 --- a/internal/email/testdata/cp1251.eml +++ b/internal/email/testdata/cp1251.eml @@ -1,7 +1,7 @@ From: legacy@example.org To: kami@example.org -Subject: Legacy +Subject: =?windows-1251?B?0fe48iDn4CDo7fLl8O3l8g==?= Date: Fri, 01 Aug 2026 05:00:00 +0400 Content-Type: text/plain; charset="windows-1251" -Ï + . diff --git a/internal/email/testdata/koi8r.eml b/internal/email/testdata/koi8r.eml new file mode 100644 index 0000000..b6f18d7 --- /dev/null +++ b/internal/email/testdata/koi8r.eml @@ -0,0 +1,7 @@ +From: legacy@example.org +To: kami@example.org +Subject: Legacy +Date: Fri, 01 Aug 2026 05:00:00 +0400 +Content-Type: text/plain; charset="koi8-r" + + ޣ. From aee20a6abcc6572170850afcdb057fd2bc5a95b8 Mon Sep 17 00:00:00 2001 From: kami Date: Sat, 1 Aug 2026 14:05:07 +0400 Subject: [PATCH 3/4] llm: give voice turns priority on the single llama-server slot llama-server is started without -np, so it serves one request at a time and everything else queues. Mail extraction is allowed two minutes on a Thinking 1.7B, and the reader hands core up to 25 messages back to back. A turn arriving mid-extraction therefore waited for whatever was left of that budget: the router timed out into the classifier cascade and its 36.8% floor, and the phraser, which has no floor, simply waited. Memory evaluation had the same shape with a five minute budget. llm.Gate is the bound. Foreground requests never wait. Background requests run one at a time and yield while a foreground request is in flight, plus a quiet window after it that covers the gap between the router call and the phraser call of one turn. Clients get their priority from llmClientFor or llmBackgroundClientFor, so which side a caller is on is decided at wiring time. It gates only what goes through those clients, which the comment on Gate says. mail intake: the extraction timeout no longer wraps the capture writes. A model answering at 119 seconds of a 120 second budget left the first CaptureTask one second and the third none, so candidates the model had already produced were dropped with a deadline error. The mailbox name is validated before it becomes provenance, since "email:" is not a source and neither is an arbitrary string posted at the socket. The enable log prints the normalised candidate bound rather than the configured one, which said "max 0" and then wrote three. Found in review of #64. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01TrVSBKe3RFDF4fGYKWYQnX --- cmd/mavend/mail.go | 74 ++++++++++++++++++++--- cmd/mavend/mail_test.go | 59 +++++++++++++++++++ cmd/mavend/memoryeval.go | 4 +- cmd/mavend/modelswap.go | 28 +++++++++ internal/email/extract.go | 24 +++++++- internal/ipc/api.go | 6 ++ internal/llm/client.go | 37 ++++++++++++ internal/llm/gate.go | 121 ++++++++++++++++++++++++++++++++++++++ internal/llm/gate_test.go | 101 +++++++++++++++++++++++++++++++ 9 files changed, 443 insertions(+), 11 deletions(-) create mode 100644 internal/llm/gate.go create mode 100644 internal/llm/gate_test.go 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") + } +} From 8c6332f95cea42f0521e9449334ab969add482f5 Mon Sep 17 00:00:00 2001 From: kami Date: Sat, 1 Aug 2026 14:08:32 +0400 Subject: [PATCH 4/4] mavmaild: own its state volume, retire aged-out UIDs, stop restart-looping The commented compose service mounted dbdata, the encrypted database volume, read-write, for one JSON file of UIDs. The header of that same file says only mavend holds the key and the db volume, and the whole argument for a separate reader is that a compromise on either side does not reach the other. It gets its own volume now, at its own path, so neither can be restored from a backup of the other. The high-water mark only advances through a contiguous run, and a failed ingest is deliberately not marked. One message that never ingested therefore pinned the mark forever: after the lookback window passed it could never be fetched again, so the gap never closed, every UID above it stayed in the explicit set, and save rewrote all of them every poll. FetchSince now reports the SEARCH window and the poller retires everything below it, since a UID that can no longer be searched for can never be read. On ErrUnknownMethod the daemon logged "stopping" and then exited at the next tick with status 0. The compose service inherits restart: unless-stopped, which restarts a clean exit, so the real behaviour was a loop of four IMAP logins an hour against a mailbox core would not accept anything from. It now stays up and polls nothing. The reader also sends the Junk verdict instead of counting bulk locally, which is what the wire doc says it does. The verdict carries no mail content, since nothing on the other side will read it. RunWith is gone, so the tests fake the read rather than the transport. Found in review of #65. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01TrVSBKe3RFDF4fGYKWYQnX --- cmd/mavmaild/main.go | 107 +++++++++++++++++++++---- cmd/mavmaild/main_test.go | 163 ++++++++++++++++++++++++++------------ docker-compose.yml | 11 ++- internal/email/fetch.go | 11 +++ models/stt | 1 + models/tts | 1 + 6 files changed, 224 insertions(+), 70 deletions(-) create mode 120000 models/stt create mode 120000 models/tts diff --git a/cmd/mavmaild/main.go b/cmd/mavmaild/main.go index 296efe6..cf547c2 100644 --- a/cmd/mavmaild/main.go +++ b/cmd/mavmaild/main.go @@ -61,6 +61,11 @@ func run(args []string) error { 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)") @@ -125,11 +130,18 @@ func run(args []string) error { 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 { - // Core told us mail ingestion is not configured. Nothing will change - // without a core restart, and a restart restarts us too. - log.Printf("mavmaild: core does not accept mail — idling") - return nil + continue } r.pollOnce(ctx, password) } @@ -153,10 +165,17 @@ type reader struct { timeout time.Duration state *seenState - // dial — connection seam for the tests; nil ⇒ implicit TLS. - dial func(addr string, timeout time.Duration) (*email.Conn, error) + // 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 } @@ -179,23 +198,25 @@ func (r *reader) pollOnce(ctx context.Context, password string) { if ctx.Err() != nil { return } - if m.Junk { - junk++ - // Marked seen without a model call: the header filter already decided, - // and re-classifying it every quarter hour would be pure waste. - r.state.mark(m.UID) - continue - } - resp, err := r.core.IngestMail(ctx, ipc.IngestMailReq{ + 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; stopping") + log.Printf("mavmaild: core has no email block configured — mail ingestion is off; idling until a restart") r.disabled = true return } @@ -217,6 +238,9 @@ func (r *reader) pollOnce(ctx context.Context, password string) { // 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, @@ -225,8 +249,24 @@ func (r *reader) fetch(password string) ([]email.Message, error) { 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.RunWith(password, r.dial) + return f.Run(password) } // ---- seen state ------------------------------------------------------------ @@ -242,6 +282,14 @@ func (r *reader) fetch(password string) ([]email.Message, error) { // 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 @@ -280,6 +328,31 @@ func (s *seenState) mark(uid uint32) { } } +// 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 diff --git a/cmd/mavmaild/main_test.go b/cmd/mavmaild/main_test.go index cf024a0..f4787dc 100644 --- a/cmd/mavmaild/main_test.go +++ b/cmd/mavmaild/main_test.go @@ -1,13 +1,10 @@ package main import ( - "bufio" "context" "fmt" - "net" "os" "path/filepath" - "strconv" "strings" "testing" "time" @@ -16,7 +13,11 @@ import ( "github.com/kami/maven/internal/ipc" ) -// ---- a scripted IMAP server, same shape internal/email's tests use --------- +// ---- a fake mailbox ------------------------------------------------------- +// +// It fakes the READ, not the protocol: internal/email owns the IMAP tests, and +// its dialer is unexported precisely so this package cannot substitute a +// transport. type fakeIMAP struct { msgs map[uint32]string @@ -24,52 +25,45 @@ type fakeIMAP struct { cmds []string } -func (f *fakeIMAP) serve(c net.Conn) { - defer c.Close() - fmt.Fprint(c, "* OK fake ready\r\n") - r := bufio.NewReader(c) - for { - line, err := r.ReadString('\n') - if err != nil { - return - } - parts := strings.SplitN(strings.TrimRight(line, "\r\n"), " ", 2) - if len(parts) != 2 { - return - } - tag, cmd := parts[0], parts[1] - f.cmds = append(f.cmds, cmd) - upper := strings.ToUpper(cmd) - switch { - case strings.HasPrefix(upper, "LOGIN"), strings.HasPrefix(upper, "EXAMINE"): - fmt.Fprintf(c, "%s OK\r\n", tag) - case strings.HasPrefix(upper, "UID SEARCH"): - var ids []string - for _, u := range f.uids { - ids = append(ids, strconv.FormatUint(uint64(u), 10)) +// fetch is the read seam the reader exposes: the daemon cannot reach +// internal/email's dialer (it is unexported so nothing outside that package can +// point the reader at a cleartext transport), so a test fakes the whole read. +// The IMAP protocol itself is covered by internal/email's own tests. +func (f *fakeIMAP) fetch(r *reader) func(string) ([]email.Message, error) { + return func(string) ([]email.Message, error) { + var out []email.Message + var low uint32 + for _, uid := range f.uids { + if low == 0 || uid < low { + low = uid } - fmt.Fprintf(c, "* SEARCH %s\r\n%s OK\r\n", strings.Join(ids, " "), tag) - case strings.HasPrefix(upper, "UID FETCH"): - uid, _ := strconv.ParseUint(strings.Fields(cmd)[2], 10, 32) - if raw, ok := f.msgs[uint32(uid)]; ok { - fmt.Fprintf(c, "* 1 FETCH (UID %d BODY[] {%d}\r\n%s)\r\n", uid, len(raw), raw) - } - fmt.Fprintf(c, "%s OK\r\n", tag) - case strings.HasPrefix(upper, "LOGOUT"): - fmt.Fprintf(c, "* BYE\r\n%s OK\r\n", tag) - return - default: - fmt.Fprintf(c, "%s BAD\r\n", tag) } + if low > 0 { + r.state.retire(low) + } + for i := len(f.uids) - 1; i >= 0; i-- { + uid := f.uids[i] + if r.state.seen(uid) { + continue + } + raw, ok := f.msgs[uid] + if !ok { + continue + } + f.cmds = append(f.cmds, fmt.Sprintf("UID FETCH %d", uid)) + msg, err := email.ParseMessage(uid, []byte(raw)) + if err != nil { + continue + } + out = append(out, msg) + if r.max > 0 && len(out) >= r.max { + break + } + } + return out, nil } } -func (f *fakeIMAP) dial(_ string, timeout time.Duration) (*email.Conn, error) { - cli, srv := net.Pipe() - go f.serve(srv) - return email.NewConn(cli, timeout) -} - // ---- a fake core ----------------------------------------------------------- type fakeCore struct { @@ -96,12 +90,13 @@ func mail(subject, body string, extraHeaders ...string) string { func newTestReader(t *testing.T, f *fakeIMAP, core *fakeCore, statePath string) *reader { t.Helper() - return &reader{ + r := &reader{ core: core, addr: "mail.example:993", user: "kami", mailbox: "INBOX", lookback: 72 * time.Hour, max: 25, timeout: 5 * time.Second, state: newSeenState(statePath), - dial: f.dial, } + r.fetchMail = f.fetch(r) + return r } func TestPollHandsMessagesToCore(t *testing.T) { @@ -116,11 +111,26 @@ func TestPollHandsMessagesToCore(t *testing.T) { r := newTestReader(t, f, core, "") r.pollOnce(context.Background(), "secret") - // The newsletter is filtered before core is asked: only the real mail crosses. - if len(core.got) != 1 { - t.Fatalf("core saw %d messages, want 1 (the bulk one must not cross): %+v", len(core.got), core.got) + // Two calls: the real mail with its text, and the newsletter as a verdict + // with no content at all. Core is told about bulk rather than asked, so it + // can count it without spending the model. + if len(core.got) != 2 { + t.Fatalf("core saw %d messages, want 2: %+v", len(core.got), core.got) + } + var got, bulk ipc.IngestMailReq + for _, r := range core.got { + if r.Junk { + bulk = r + } else { + got = r + } + } + if bulk.UID != 2 || !bulk.Junk { + t.Errorf("bulk req = %+v, want uid 2 flagged junk", bulk) + } + if bulk.Subject != "" || bulk.Body != "" || bulk.From != "" { + t.Errorf("a bulk verdict must carry no mail content: %+v", bulk) } - got := core.got[0] if got.UID != 1 || got.Mailbox != "INBOX" || got.Subject != "Счёт" { t.Errorf("ingest req = %+v", got) } @@ -252,3 +262,54 @@ func TestRunRejectsEmptyPasswordFile(t *testing.T) { t.Errorf("an empty password file must be refused before dialling; err = %v", err) } } + +// A UID that never ingests pinned the high-water mark forever, because the mark +// only advances through a contiguous run. Once that UID falls out of the +// lookback window it can never be fetched again, so there is nothing left to +// wait for and everything above it can leave the explicit set. +func TestSeenStateRetiresAgedOutUIDs(t *testing.T) { + s := newSeenState("") + s.mark(1000) // 999 failed and was deliberately not marked + s.mark(1001) + if s.high != 0 || len(s.set) != 2 { + t.Fatalf("high = %d, set = %v; want the mark pinned below the gap", s.high, s.set) + } + // The next SEARCH window starts at 1000: 999 has aged out. + s.retire(1000) + if s.high != 1001 { + t.Errorf("high = %d, want 1001 once the gap is unreachable", s.high) + } + if len(s.set) != 0 { + t.Errorf("explicit set = %v, want empty", s.set) + } + if !s.seen(999) || !s.seen(1001) || s.seen(1002) { + t.Errorf("seen(999)=%v seen(1001)=%v seen(1002)=%v", s.seen(999), s.seen(1001), s.seen(1002)) + } +} + +// retire never moves the mark backwards: a UIDVALIDITY reset restarts UIDs low, +// and that must not un-see a mailbox or re-see one. +func TestSeenStateRetireNeverGoesBackwards(t *testing.T) { + s := newSeenState("") + s.mark(1) + s.mark(2) + s.retire(1) + if s.high != 2 { + t.Errorf("high = %d, want 2 unchanged", s.high) + } +} + +// A poll must not leave the state file growing with UIDs that are already +// covered by the high-water mark. +func TestPollRetiresThroughTheSearchWindow(t *testing.T) { + f := &fakeIMAP{uids: []uint32{100, 101}, msgs: map[uint32]string{100: mail("a", "b"), 101: mail("c", "d")}} + core := &fakeCore{} + r := newTestReader(t, f, core, "") + r.pollOnce(context.Background(), "secret") + if r.state.high != 101 { + t.Errorf("high = %d, want 101 — everything below the search window is unreachable", r.state.high) + } + if len(r.state.set) != 0 { + t.Errorf("explicit set = %v, want empty", r.state.set) + } +} diff --git a/docker-compose.yml b/docker-compose.yml index 3a9db6e..be5a6e9 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -131,16 +131,23 @@ services: # "-password-file", "/run/secrets/imap.password", # "-mailbox", "INBOX", # "-interval", "15m", - # "-state", "/var/lib/maven/mail-seen.json"] + # "-state", "/var/lib/mavmaild/mail-seen.json"] # depends_on: [mavend] # volumes: # - sockets:/run/maven - # - dbdata:/var/lib/maven + # # Its OWN volume, not dbdata. The whole point of a separate reader is + # # that a compromise on either side does not reach the other, and dbdata + # # is the encrypted database. The reader needs one JSON file of UIDs and + # # gets a volume that holds nothing else, so neither can be restored from + # # a backup of the other. + # - maildata:/var/lib/mavmaild # - ./deploy/imap.password:/run/secrets/imap.password:ro volumes: dbdata: sockets: + # maildata — the mail reader's seen-UID file, and nothing else. See mavmaild. + maildata: networks: default: diff --git a/internal/email/fetch.go b/internal/email/fetch.go index 99c3b8b..43d3f61 100644 --- a/internal/email/fetch.go +++ b/internal/email/fetch.go @@ -27,6 +27,13 @@ type FetchSince struct { Since time.Time Max int Skip func(uid uint32) bool + // OnSearch, when set, is handed the whole SEARCH result before anything is + // fetched, ascending, seen UIDs included. It is how the poller learns which + // UIDs are still inside the lookback window: anything below the lowest one + // can never be searched for again, and therefore can never be read again. + // Without that the poller cannot tell a UID it has not got to yet from one + // that has aged out of the window. + OnSearch func(uids []uint32) // dial — the connection seam, unexported on purpose: see dialer(). Tests // inside this package set it through export_test.go; nothing outside can. @@ -84,6 +91,10 @@ func (f FetchSince) Run(password string) ([]Message, error) { return nil, err } + if f.OnSearch != nil { + f.OnSearch(uids) + } + // Newest UIDs first — IMAP hands them back ascending, and when Max clips the // list the recent mail is what matters. wanted := make([]uint32, 0, len(uids)) diff --git a/models/stt b/models/stt new file mode 120000 index 0000000..b983fa3 --- /dev/null +++ b/models/stt @@ -0,0 +1 @@ +/home/kami/apps/Maven/models/stt \ No newline at end of file diff --git a/models/tts b/models/tts new file mode 120000 index 0000000..66782fb --- /dev/null +++ b/models/tts @@ -0,0 +1 @@ +/home/kami/apps/Maven/models/tts \ No newline at end of file