Merge branch 'fix/g06' into fix/integrated

# Conflicts:
#	cmd/mavend/memoryeval.go

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}
 }
 
@@ -88,6 +129,13 @@ func newMailIntake(st *store.Store, phr phraser.Phraser, cfg *config.Config, bus
 // to the point, a task he already marked done is not re-proposed the next time
 // the same unread message is read again.
 func (m *mailIntake) ingest(ctx context.Context, req ipc.IngestMailReq) (ipc.IngestMailResp, error) {
+	// The mailbox name becomes provenance ("email:INBOX"), and the source
+	// vocabulary is what the loop's rules trust. An empty name gave "email:" and
+	// an arbitrary string gave an arbitrary source under that namespace.
+	mailbox, err := validMailbox(req.Mailbox)
+	if err != nil {
+		return ipc.IngestMailResp{}, err
+	}
 	msg := email.Message{
 		UID:     req.UID,
 		From:    req.From,
@@ -100,9 +148,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.
@@ -112,7 +166,13 @@ func (m *mailIntake) ingest(ctx context.Context, req ipc.IngestMailReq) (ipc.Ing
 		return ipc.IngestMailResp{}, nil
 	}
 
-	source := email.SourcePrefix + req.Mailbox
+	// A fresh budget for the writes, derived from the caller's context rather
+	// than from the extraction's. Encrypted-store writes are fast; what this
+	// bounds is a stuck store, not the model.
+	ctx, cancel = context.WithTimeout(ctx, captureTimeout)
+	defer cancel()
+
+	source := email.SourcePrefix + mailbox
 	evidence := truncateRunes(req.Subject, evidenceMaxChars)
 	now := m.now()
 	var resp ipc.IngestMailResp

cmd/mavend/mail_test.go

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

cmd/mavend/memoryeval.go

@@ -25,11 +25,14 @@ import (
 //
 // It used to be five minutes, on the grounds that nobody waits for the answer.
 // Nobody waits for the evaluation, but there is ONE resident model behind one
-// llama-server, so a voice turn that arrives mid-evaluation waits behind it:
-// five minutes of evaluation is five minutes of a mute assistant. Sixty seconds
-// is long enough for a Thinking model on this prompt and short enough that the
-// worst collision is one turn answered late rather than a turn abandoned. An
-// evaluation cut off here costs nothing: it is retried at the next interval.
+// llama-server, so a voice turn arriving mid-evaluation waited behind it: five
+// minutes of evaluation was five minutes of a mute assistant.
+//
+// The background client now yields the slot while a turn is in flight, so the
+// collision is handled where it belongs and this is a prompt budget again.
+// Sixty seconds is long enough for a Thinking model here, and an evaluation cut
+// off costs nothing, because it is retried at the next interval. Raise it if
+// observations start truncating.
 const memoryEvalTimeout = 60 * time.Second
 
 // memoryEvalWorker — ticker + evaluator.
@@ -58,7 +61,9 @@ func newMemoryEvalWorker(st *store.Store, phr phraser.Phraser, cfg *config.Confi
 	if interval <= 0 {
 		interval = config.DefaultMemoryEvalInterval
 	}
-	client := llmClientFor(lp, memoryEvalTimeout)
+	// 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, memoryEvalTimeout)
 	ev := memeval.NewEvaluator(st, st, client, memeval.Config{
 		MaxItems:      cfg.MemoryEval.MaxItems,
 		MinConfidence: cfg.MemoryEval.MinConfidence,

cmd/mavend/modelswap.go

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

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

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)
+	}
+}