Run the reader's batches: paced, judged, written as entries, unhealthy on failure (issue 1520)

modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go

@@ -0,0 +1,252 @@
+package main
+
+// Running the reader's batches (novox/hq ADR 0459 §5, §14, §15; issue 1520): pacing batch starts at least
+// BatchGap apart, judging one batch's question (question.go, issue 1427) with claude_code_judge and collecting
+// a pending answer with claude_code_judged (mirrors mesh-knowledge's own judgementOver, cmd/mesh-knowledge/main.go),
+// and what a batch's answer does to the store: one entry per mail and any new mail group on success, nothing
+// written and the reader made unhealthy on a refused, timed-out or out-of-shape judgement (1519's state).
+//
+// mesh-mail-reader never asks to send, move, flag or delete (novox/hq ADR 0459 §11): every bus call in this
+// file goes through one Ask func, so a test can record every key it was given and prove that (1520/AC5).
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+)
+
+// BatchGap is how long must pass between the start of one batch and the next (ADR 0459 §15, 1520/AC1).
+const BatchGap = 160 * time.Second
+
+// JudgeWithin is the bound every judgement asks: never above 40 seconds (ADR 0459 §14, 1520/AC2).
+const JudgeWithin = 40 * time.Second
+
+// Ask is the one primitive every bus call in this file goes through — stdio.Ask in the running module
+// (main.go), a recording stand-in in a test (1520/AC5).
+type Ask func(key string, body map[string]any) (json.RawMessage, error)
+
+// ReadTextFunc fetches one mail's text, never kept beyond the call that asked for it.
+type ReadTextFunc func(account, folder string, uid uint32) (string, error)
+
+// NewReadText is ReadTextFunc over seat:mail-store.read (ADR 0459 §3): the same seat mesh-mail-reader's
+// manifest already invokes (1426/AC2).
+func NewReadText(ask Ask) ReadTextFunc {
+	return func(account, folder string, uid uint32) (string, error) {
+		raw, err := ask("seat:mail-store.read", map[string]any{"account": account, "folder": folder, "uid": int(uid)})
+		if err != nil {
+			return "", err
+		}
+		var out struct {
+			Text string `json:"text"`
+		}
+		if err := json.Unmarshal(raw, &out); err != nil {
+			return "", fmt.Errorf("mail-store.read's answer is not a message: %.200s", raw)
+		}
+		return out.Text, nil
+	}
+}
+
+// JudgeFunc asks one batch's question and answers its outcome: the raw answer and OutcomeOK on success, or no
+// answer and the outcome that says why (1520/AC4).
+type JudgeFunc func(question string, schema map[string]any) (answer json.RawMessage, outcome HealthOutcome, why string)
+
+// NewJudge is JudgeFunc over claude-code.claude_code_judge on this reader's own machine (the lead's decision
+// at this milestone's prepare step: node is MESH_NODE), collecting a pending answer with claude_code_judged
+// and its ticket until it answers or JudgeWithin plus a minute's grace has passed (1520/AC2) — the same shape
+// mesh-knowledge's own judgementOver polls (cmd/mesh-knowledge/main.go). A bus call that itself errors (the
+// judge module refused the ask outright, such as "holds no licence") is OutcomeRefused; an answer that keeps no
+// shape ParseAnswer or this loop can read is OutcomeOutOfShape; running past the deadline is OutcomeTimedOut.
+func NewJudge(ask Ask, node string, now func() time.Time) JudgeFunc {
+	key := "claude-code.claude_code_judge"
+	if node != "" {
+		key += "@" + node
+	}
+	return func(question string, schema map[string]any) (json.RawMessage, HealthOutcome, string) {
+		raw, err := ask(key, map[string]any{"question": question, "schema": schema, "within": int(JudgeWithin / time.Second)})
+		deadline := now().Add(JudgeWithin + time.Minute)
+		for {
+			if err != nil {
+				return nil, OutcomeRefused, err.Error()
+			}
+			var out struct {
+				Answer  json.RawMessage `json:"answer"`
+				Node    string          `json:"node"`
+				Ticket  string          `json:"ticket"`
+				Pending bool            `json:"pending"`
+			}
+			if jsonErr := json.Unmarshal(raw, &out); jsonErr != nil {
+				return nil, OutcomeOutOfShape, fmt.Sprintf("the judge's answer is not a judgement: %.200s", raw)
+			}
+			if !out.Pending {
+				if len(out.Answer) == 0 {
+					return nil, OutcomeOutOfShape, fmt.Sprintf("the judge's answer is not a judgement: %.200s", raw)
+				}
+				return out.Answer, OutcomeOK, ""
+			}
+			if out.Ticket == "" || out.Node == "" {
+				return nil, OutcomeOutOfShape, "the judge answered pending with no ticket or node"
+			}
+			if now().After(deadline) {
+				return nil, OutcomeTimedOut, fmt.Sprintf("no answer within %s", JudgeWithin)
+			}
+			raw, err = ask("claude-code.claude_code_judged@"+out.Node, map[string]any{"ticket": out.Ticket})
+		}
+	}
+}
+
+// nextBatchWait is how long to wait before the next batch may start (1520/AC1): zero once BatchGap has
+// passed since the last start (or no batch has started yet), else the remainder.
+func nextBatchWait(lastStart, now time.Time) time.Duration {
+	if lastStart.IsZero() {
+		return 0
+	}
+	elapsed := now.Sub(lastStart)
+	if elapsed >= BatchGap {
+		return 0
+	}
+	return BatchGap - elapsed
+}
+
+func senderOf(from []Person) (name, address string) {
+	if len(from) == 0 {
+		return "", ""
+	}
+	return from[0].Name, from[0].Address
+}
+
+func groupSummariesFrom(ctx context.Context, store Store) ([]GroupSummary, error) {
+	groups, err := store.Groups(ctx)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]GroupSummary, len(groups))
+	for i, g := range groups {
+		out[i] = GroupSummary{Label: g.Label, Count: g.Count}
+	}
+	return out, nil
+}
+
+// recordOutcome folds one judgement's outcome into the reader's kept health (1519, 1520/AC4): writes no
+// entry, and leaves whatever the caller already left alone.
+func recordOutcome(ctx context.Context, store Store, outcome HealthOutcome, why string, now time.Time) error {
+	current, err := store.ReaderHealth(ctx)
+	if err != nil {
+		return err
+	}
+	return store.SetReaderHealth(ctx, RecordJudgement(current, outcome, why, now))
+}
+
+// RunOneBatch takes at most MaxMailsPerBatch of the waiting mails given (already in arrival order — the
+// caller's to select, so a test can hand it an exact slice), builds the question (question.go) from them, the
+// existing mail groups and the applicable reading lessons, judges it, and applies the answer:
+//
+//   - on success (1520/AC3): one mail entry per mail, any new mail group found-or-created by its label, and
+//     each mail leaves the waiting queue;
+//   - on a refused, timed-out or out-of-shape judgement (1520/AC4): no entry is written, every given mail
+//     stays waiting, in the same order, and the reader's health (1519) is set unhealthy with the outcome and
+//     why; the next successful batch sets it healthy again.
+//
+// mesh-mail-reader never asks to send, move, flag or delete (ADR 0459 §11): this function asks only readText
+// (seat:mail-store.read) and judge (claude-code.claude_code_judge, claude_code_judged) — 1520/AC5 is proven by
+// building both from a recording Ask and checking what it recorded, not by anything this function does itself.
+func RunOneBatch(ctx context.Context, store Store, readText ReadTextFunc, judge JudgeFunc, waiting []WaitingMail, now func() time.Time) error {
+	if len(waiting) > MaxMailsPerBatch {
+		waiting = waiting[:MaxMailsPerBatch]
+	}
+	if len(waiting) == 0 {
+		return nil
+	}
+	lessons, err := store.Lessons(ctx)
+	if err != nil {
+		return err
+	}
+	groups, err := groupSummariesFrom(ctx, store)
+	if err != nil {
+		return err
+	}
+	mails := make([]QuestionMail, len(waiting))
+	ids := make([]int64, len(waiting))
+	for i, w := range waiting {
+		text, err := readText(w.Account, w.Folder, w.UID)
+		if err != nil {
+			// mail-store.read itself failing stops the batch exactly as a refused judgement would: the mails
+			// stay waiting, and the reader is made unhealthy with the cause.
+			return recordOutcome(ctx, store, OutcomeRefused, "reading the mail's text: "+err.Error(), now())
+		}
+		mails[i] = QuestionMail{ID: w.Seq, Account: w.Account, Folder: w.Folder, Date: w.Date, From: w.From, Subject: w.Subject, Flags: w.Flags, Text: text}
+		ids[i] = w.Seq
+	}
+	question := BuildQuestion(mails, groups, lessons)
+	raw, outcome, why := judge(question.Text, AnswerSchema())
+	if outcome != OutcomeOK {
+		return recordOutcome(ctx, store, outcome, why, now())
+	}
+	answer, err := ParseAnswer(raw, ids)
+	if err != nil {
+		return recordOutcome(ctx, store, OutcomeOutOfShape, err.Error(), now())
+	}
+	judgedAt := now()
+	for _, w := range waiting {
+		j := answer.Mails[w.Seq]
+		var groupID int64
+		if j.Group != "" {
+			g, err := store.FindOrCreateGroup(ctx, j.Group)
+			if err != nil {
+				return err
+			}
+			groupID = g.ID
+		}
+		var due time.Time
+		if j.DueDate != "" {
+			due, _ = time.Parse("2006-01-02", j.DueDate) // ParseAnswer already proved this parses
+		}
+		name, address := senderOf(w.From)
+		e := Entry{
+			Account: w.Account, MessageID: w.MessageID, Folder: w.Folder, Date: w.Date,
+			SenderName: name, SenderAddress: address, Subject: w.Subject,
+			Group: groupID, Summary: j.Summary, Mark: j.Mark, Why: j.Why, DueDate: due, Judged: judgedAt,
+		}
+		if _, err := store.AddEntry(ctx, e); err != nil {
+			return err
+		}
+		if err := store.TakeWaiting(ctx, w.Seq); err != nil {
+			return err
+		}
+	}
+	return recordOutcome(ctx, store, OutcomeOK, "", judgedAt)
+}
+
+// RunBatches is the loop the running module drives (main.go): pace batch starts at least BatchGap apart
+// (nextBatchWait), take at most MaxMailsPerBatch of the waiting mails in arrival order, and run one batch at a
+// time, forever, until ctx is done. Not unit-tested itself beyond compiling (go vet): 1520/AC1's pacing and
+// AC2-AC5's one-batch behaviour are each proven directly, on nextBatchWait and RunOneBatch, in batch_test.go.
+func RunBatches(ctx context.Context, store Store, readText ReadTextFunc, judge JudgeFunc, now func() time.Time, sleep func(time.Duration), logf func(string, ...any)) {
+	var lastStart time.Time
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		if wait := nextBatchWait(lastStart, now()); wait > 0 {
+			sleep(wait)
+			continue
+		}
+		waiting, err := store.Waiting(ctx)
+		if err != nil {
+			logf("[mesh-mail-reader] the waiting mails could not be read: %v", err)
+			sleep(time.Second)
+			continue
+		}
+		if len(waiting) == 0 {
+			sleep(time.Second)
+			continue
+		}
+		if len(waiting) > MaxMailsPerBatch {
+			waiting = waiting[:MaxMailsPerBatch]
+		}
+		lastStart = now()
+		if err := RunOneBatch(ctx, store, readText, judge, waiting, now); err != nil {
+			logf("[mesh-mail-reader] a batch failed: %v", err)
+		}
+	}
+}

modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_test.go

@@ -0,0 +1,305 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+)
+
+func seedWaiting(t *testing.T, store Store, n int) {
+	t.Helper()
+	for i := 0; i < n; i++ {
+		err := store.AddWaiting(context.Background(), WaitingMail{
+			Account: "bob", Folder: "INBOX", MessageID: fmt.Sprintf("<%d@example.org>", i),
+			UIDValidity: 1, UID: uint32(i + 1), Date: time.Date(2026, 10, 20, 10, 0, 0, 0, time.UTC),
+			From: []Person{{Name: "Alice", Address: "alice@example.org"}}, Subject: "hello",
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func noopReadText(string, string, uint32) (string, error) { return "the mail's text", nil }
+
+// judgeAllNeither is a JudgeFunc that marks every mail of the given waiting list "neither", no group, no
+// due date — a trivial, always-successful stand-in for the tests that only care about pacing or plumbing,
+// not the judge's own polling (that is NewJudge's own test).
+func judgeAllNeither(waiting []WaitingMail) JudgeFunc {
+	return func(string, map[string]any) (json.RawMessage, HealthOutcome, string) {
+		mails := make([]map[string]any, len(waiting))
+		for i, w := range waiting {
+			mails[i] = map[string]any{"id": w.Seq, "mark": "neither"}
+		}
+		raw, _ := json.Marshal(map[string]any{"mails": mails})
+		return raw, OutcomeOK, ""
+	}
+}
+
+// 1520/AC1 (the pure pacing math): zero once BatchGap has passed since the last start, or no batch has
+// started yet; the remainder otherwise.
+func TestNextBatchWait(t *testing.T) {
+	start := time.Date(2026, 10, 20, 10, 0, 0, 0, time.UTC)
+	if w := nextBatchWait(time.Time{}, start); w != 0 {
+		t.Fatalf("no batch yet: wait = %s, want 0", w)
+	}
+	if w := nextBatchWait(start, start.Add(100*time.Second)); w != 60*time.Second {
+		t.Fatalf("100s after: wait = %s, want 60s", w)
+	}
+	if w := nextBatchWait(start, start.Add(BatchGap)); w != 0 {
+		t.Fatalf("exactly BatchGap after: wait = %s, want 0 (one waiting mail starts a batch at once)", w)
+	}
+	if w := nextBatchWait(start, start.Add(BatchGap+time.Second)); w != 0 {
+		t.Fatalf("past BatchGap: wait = %s, want 0", w)
+	}
+}
+
+// 1520/AC1: with 30 mails waiting, three batches of 10 start at least 160s apart (a test clock: the batch
+// loop here is driven by hand, jumping straight to each batch's own allowed start, as nextBatchWait says to).
+func TestThirtyWaitingMailsRunAsThreeBatchesAtLeast160sApart(t *testing.T) {
+	ctx := context.Background()
+	store := newMemStore()
+	seedWaiting(t, store, 30)
+
+	now := time.Date(2026, 10, 20, 10, 0, 0, 0, time.UTC)
+	var lastStart time.Time
+	var starts []time.Time
+	batches := 0
+	for {
+		all, err := store.Waiting(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(all) == 0 {
+			break
+		}
+		if wait := nextBatchWait(lastStart, now); wait > 0 {
+			now = now.Add(wait)
+		}
+		batch := all
+		if len(batch) > MaxMailsPerBatch {
+			batch = batch[:MaxMailsPerBatch]
+		}
+		nowFunc := func() time.Time { return now }
+		if err := RunOneBatch(ctx, store, noopReadText, judgeAllNeither(batch), batch, nowFunc); err != nil {
+			t.Fatal(err)
+		}
+		starts = append(starts, now)
+		lastStart = now
+		batches++
+		now = now.Add(time.Second) // the batch itself takes a little time
+	}
+	if batches != 3 {
+		t.Fatalf("ran %d batches, want 3 (30 mails / 10 per batch)", batches)
+	}
+	for i := 1; i < len(starts); i++ {
+		if gap := starts[i].Sub(starts[i-1]); gap < BatchGap {
+			t.Fatalf("batch %d started %s after the previous, want at least %s", i, gap, BatchGap)
+		}
+	}
+	if left, _ := store.Waiting(ctx); len(left) != 0 {
+		t.Fatalf("%d mails still waiting, want none", len(left))
+	}
+}
+
+// 1520/AC2: within is never above 40; a pending answer is collected with claude_code_judged and its ticket
+// until answered or the 40s are spent.
+func TestNewJudgeCollectsAPendingAnswer(t *testing.T) {
+	var sawWithin int
+	var sawTicket string
+	collected := 0
+	ask := func(key string, body map[string]any) (json.RawMessage, error) {
+		if key == "claude-code.claude_code_judge@novox" {
+			sawWithin = body["within"].(int)
+			return json.Marshal(map[string]any{"pending": true, "ticket": "tk-1", "node": "novox"})
+		}
+		if key == "claude-code.claude_code_judged@novox" {
+			sawTicket = body["ticket"].(string)
+			collected++
+			if collected < 2 {
+				return json.Marshal(map[string]any{"pending": true, "ticket": "tk-1", "node": "novox"})
+			}
+			return json.Marshal(map[string]any{"answer": json.RawMessage(`{"mails":[{"id":1,"mark":"neither"}]}`)})
+		}
+		return nil, fmt.Errorf("unexpected ask %s", key)
+	}
+	now := time.Now()
+	judge := NewJudge(ask, "novox", func() time.Time { return now })
+	raw, outcome, why := judge("the question", AnswerSchema())
+	if outcome != OutcomeOK {
+		t.Fatalf("outcome = %q, why = %q, want OutcomeOK", outcome, why)
+	}
+	if sawWithin > 40 {
+		t.Fatalf("within = %d, want at most 40", sawWithin)
+	}
+	if sawTicket != "tk-1" {
+		t.Fatalf("ticket = %q, want tk-1", sawTicket)
+	}
+	if collected != 2 {
+		t.Fatalf("claude_code_judged was asked %d time(s), want 2 (collected until answered)", collected)
+	}
+	if len(raw) == 0 {
+		t.Fatal("no answer")
+	}
+}
+
+// 1520/AC2: past the within plus a minute's grace, a still-pending judgement is OutcomeTimedOut. NewJudge
+// computes its own deadline from now() at the moment it is asked, so the clock is advanced from inside the
+// stand-in ask — simulating collecting taking longer than the deadline allows — never by setting it ahead of
+// time, which a fixed deadline would not then exceed.
+func TestNewJudgeTimesOut(t *testing.T) {
+	now := time.Now()
+	first := true
+	ask := func(key string, body map[string]any) (json.RawMessage, error) {
+		if first {
+			first = false
+		} else {
+			now = now.Add(JudgeWithin + 2*time.Minute) // collecting again takes "too long"
+		}
+		return json.Marshal(map[string]any{"pending": true, "ticket": "tk-1", "node": "novox"})
+	}
+	judge := NewJudge(ask, "novox", func() time.Time { return now })
+	_, outcome, why := judge("the question", AnswerSchema())
+	if outcome != OutcomeTimedOut {
+		t.Fatalf("outcome = %q, why = %q, want OutcomeTimedOut", outcome, why)
+	}
+}
+
+// 1520/AC2: the agent module's own refusal (an error from the ask itself) is OutcomeRefused.
+func TestNewJudgeRefused(t *testing.T) {
+	ask := func(string, map[string]any) (json.RawMessage, error) {
+		return nil, fmt.Errorf("holds no licence")
+	}
+	judge := NewJudge(ask, "novox", time.Now)
+	_, outcome, why := judge("the question", AnswerSchema())
+	if outcome != OutcomeRefused || !strings.Contains(why, "licence") {
+		t.Fatalf("outcome = %q, why = %q, want OutcomeRefused naming the licence", outcome, why)
+	}
+}
+
+// 1520/AC3: an answered batch writes one mail entry per mail and each new mail group, and the waiting mails
+// leave the queue.
+func TestRunOneBatchWritesEntriesAndGroups(t *testing.T) {
+	ctx := context.Background()
+	store := newMemStore()
+	seedWaiting(t, store, 2)
+	waiting, _ := store.Waiting(ctx)
+
+	judge := func(string, map[string]any) (json.RawMessage, HealthOutcome, string) {
+		raw, _ := json.Marshal(map[string]any{"mails": []map[string]any{
+			{"id": waiting[0].Seq, "group": "newsletters", "summary": "s1", "mark": "neither"},
+			{"id": waiting[1].Seq, "group": "newsletters", "summary": "s2", "mark": "matters", "why": "pay by", "due-date": "2026-11-01"},
+		}})
+		return raw, OutcomeOK, ""
+	}
+	now := time.Date(2026, 10, 20, 11, 0, 0, 0, time.UTC)
+	if err := RunOneBatch(ctx, store, noopReadText, judge, waiting, func() time.Time { return now }); err != nil {
+		t.Fatal(err)
+	}
+	entries, _ := store.Entries(ctx)
+	if len(entries) != 2 {
+		t.Fatalf("entries = %v, want 2", entries)
+	}
+	groups, _ := store.Groups(ctx)
+	if len(groups) != 1 || groups[0].Label != "newsletters" || groups[0].Count != 2 {
+		t.Fatalf("groups = %+v, want one group 'newsletters' holding 2", groups)
+	}
+	left, _ := store.Waiting(ctx)
+	if len(left) != 0 {
+		t.Fatalf("%d mails still waiting, want none", len(left))
+	}
+	h, _ := store.ReaderHealth(ctx)
+	if !h.Healthy {
+		t.Fatalf("health = %+v, want healthy after a successful batch", h)
+	}
+}
+
+// 1520/AC4: a refused, timed-out or out-of-shape judgement writes no entry, keeps the mails waiting in
+// order, and sets the reader unhealthy with its cause; the next good batch sets it healthy.
+func TestRunOneBatchFailureKeepsWaitingAndSetsUnhealthy(t *testing.T) {
+	cases := []struct {
+		name    string
+		judge   JudgeFunc
+		outcome HealthOutcome
+	}{
+		{"refused", func(string, map[string]any) (json.RawMessage, HealthOutcome, string) {
+			return nil, OutcomeRefused, "holds no licence"
+		}, OutcomeRefused},
+		{"timed out", func(string, map[string]any) (json.RawMessage, HealthOutcome, string) {
+			return nil, OutcomeTimedOut, "no answer in 40s"
+		}, OutcomeTimedOut},
+		{"out of shape", func(string, map[string]any) (json.RawMessage, HealthOutcome, string) {
+			return json.RawMessage(`{"mails":[]}`), OutcomeOK, "" // parsed below: this is the OK branch that then fails ParseAnswer
+		}, OutcomeOutOfShape},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			ctx := context.Background()
+			store := newMemStore()
+			seedWaiting(t, store, 2)
+			waiting, _ := store.Waiting(ctx)
+			now := time.Date(2026, 10, 20, 9, 0, 0, 0, time.UTC)
+			if err := RunOneBatch(ctx, store, noopReadText, c.judge, waiting, func() time.Time { return now }); err != nil {
+				t.Fatal(err)
+			}
+			if entries, _ := store.Entries(ctx); len(entries) != 0 {
+				t.Fatalf("entries = %v, want none", entries)
+			}
+			left, _ := store.Waiting(ctx)
+			if len(left) != 2 || left[0].Seq != waiting[0].Seq || left[1].Seq != waiting[1].Seq {
+				t.Fatalf("waiting = %v, want the same 2 mails, in the same order", left)
+			}
+			h, _ := store.ReaderHealth(ctx)
+			if h.Healthy || h.Cause != c.outcome {
+				t.Fatalf("health = %+v, want unhealthy with cause %q", h, c.outcome)
+			}
+
+			// The next good batch sets it healthy again.
+			later := now.Add(time.Hour)
+			if err := RunOneBatch(ctx, store, noopReadText, judgeAllNeither(waiting), waiting, func() time.Time { return later }); err != nil {
+				t.Fatal(err)
+			}
+			h, _ = store.ReaderHealth(ctx)
+			if !h.Healthy {
+				t.Fatalf("health = %+v, want healthy after the next good batch", h)
+			}
+		})
+	}
+}
+
+// 1520/AC5: the reader never asks mail-store.send, send-status or anything that moves, flags or deletes — a
+// stand-in records every ask RunOneBatch makes, start to finish.
+func TestRunOneBatchNeverAsksToSendMoveFlagOrDelete(t *testing.T) {
+	ctx := context.Background()
+	store := newMemStore()
+	seedWaiting(t, store, 1)
+	waiting, _ := store.Waiting(ctx)
+
+	var asked []string
+	recordingAsk := func(key string, body map[string]any) (json.RawMessage, error) {
+		asked = append(asked, key)
+		if strings.HasPrefix(key, "claude-code.claude_code_judge") {
+			raw, _ := json.Marshal(map[string]any{"mails": []map[string]any{{"id": waiting[0].Seq, "mark": "neither"}}})
+			return json.Marshal(map[string]any{"answer": raw})
+		}
+		return json.Marshal(map[string]any{"text": "hi"})
+	}
+	readText := NewReadText(recordingAsk)
+	judge := NewJudge(recordingAsk, "novox", time.Now)
+	if err := RunOneBatch(ctx, store, readText, judge, waiting, time.Now); err != nil {
+		t.Fatal(err)
+	}
+	if len(asked) == 0 {
+		t.Fatal("RunOneBatch asked nothing at all")
+	}
+	for _, key := range asked {
+		for _, forbidden := range []string{"send", "move", "flag", "delete"} {
+			if strings.Contains(strings.ToLower(key), forbidden) {
+				t.Fatalf("asked %q, which names %q — the reader must never send, move, flag or delete", key, forbidden)
+			}
+		}
+	}
+}

modules/mesh-mail-reader/cmd/mesh-mail-reader/kit_test.go

@@ -17,6 +17,13 @@ type memStore struct {
 	nextSeq  int64
 	failNext error // AddWaiting fails once with this, then clears, so a test can prove a failure is returned
 	health   ReaderHealth
+
+	entries    []Entry
+	nextEntry  int64
+	groups     []Group
+	nextGroup  int64
+	lessons    []Lesson
+	nextLesson int64
 }
 
 func newMemStore() *memStore {
@@ -75,4 +82,104 @@ func (s *memStore) SetReaderHealth(_ context.Context, h ReaderHealth) error {
 	return nil
 }
 
+func (s *memStore) AddEntry(_ context.Context, e Entry) (int64, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.nextEntry++
+	e.ID = s.nextEntry
+	s.entries = append(s.entries, e)
+	return e.ID, nil
+}
+
+func (s *memStore) UpdateEntry(_ context.Context, e Entry) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for i, existing := range s.entries {
+		if existing.ID == e.ID {
+			s.entries[i] = e
+			return nil
+		}
+	}
+	return ErrNotFound
+}
+
+func (s *memStore) Entry(_ context.Context, id int64) (Entry, bool, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, e := range s.entries {
+		if e.ID == id {
+			return e, true, nil
+		}
+	}
+	return Entry{}, false, nil
+}
+
+func (s *memStore) Entries(context.Context) ([]Entry, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make([]Entry, len(s.entries))
+	copy(out, s.entries)
+	return out, nil
+}
+
+func (s *memStore) FindOrCreateGroup(_ context.Context, label string) (Group, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, g := range s.groups {
+		if g.Label == label {
+			return g, nil
+		}
+	}
+	s.nextGroup++
+	g := Group{ID: s.nextGroup, Label: label}
+	s.groups = append(s.groups, g)
+	return g, nil
+}
+
+func (s *memStore) Groups(context.Context) ([]Group, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	counts := map[int64]int{}
+	for _, e := range s.entries {
+		if e.Group != 0 {
+			counts[e.Group]++
+		}
+	}
+	out := make([]Group, len(s.groups))
+	for i, g := range s.groups {
+		g.Count = counts[g.ID]
+		out[i] = g
+	}
+	return out, nil
+}
+
+func (s *memStore) AddLesson(_ context.Context, l Lesson) (int64, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.nextLesson++
+	l.ID = s.nextLesson
+	s.lessons = append(s.lessons, l)
+	return l.ID, nil
+}
+
+func (s *memStore) Lessons(context.Context) ([]Lesson, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make([]Lesson, len(s.lessons))
+	copy(out, s.lessons)
+	return out, nil
+}
+
+func (s *memStore) UndoLesson(_ context.Context, id int64) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for i, l := range s.lessons {
+		if l.ID == id {
+			s.lessons[i].Undone = true
+			return nil
+		}
+	}
+	return ErrNotFound
+}
+
 var errStoreFake = errors.New("the store refused (fake, for a test)")

modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go

@@ -1,12 +1,13 @@
 // mesh-mail-reader: the default claimant of mail-reading (novox/hq ADR 0459 §1), built from mesh-mail's own
 // source, like mesh-mail-web and mesh-mail-notifier. One instance, held once for the mesh, on the machine
 // mesh-mail runs on. It hears mesh-mail's added events under its own runs-as account (ADR 0459 §3, issue
-// 1518), judges a batch of them at a time (1427, 1520 — not yet built), and serves the six mail-reading
-// verbs (entries, groups, correct, lessons, undo, status) and its own health tool, mail_reader_health
-// (1519). It reads mail; it moves, flags, deletes and sends nothing (novox/hq ADR 0459 §11).
+// 1518), judges a batch of them at a time (1427, 1520), and serves the six mail-reading verbs (entries,
+// groups, correct, lessons, undo, status) and its own health tool, mail_reader_health (1519). It reads mail;
+// it moves, flags, deletes and sends nothing (novox/hq ADR 0459 §11).
 //
-//	mesh-mail-reader            the bundle the node's runtime launches: the hearing loop (hear.go) and the
-//	                            six mail-reading verbs plus mail_reader_health, served over stdio (tools.go)
+//	mesh-mail-reader            the bundle the node's runtime launches: the hearing loop (hear.go), the
+//	                            batch loop (batch.go) and the six mail-reading verbs plus mail_reader_health,
+//	                            served over stdio (tools.go)
 //	mesh-mail-reader prepare    the preparation step: the index's schema (store.go, pgstore.go) brought to
 //	                            this version's (issue 1517)
 //
@@ -15,6 +16,7 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"os"
 	"time"
@@ -38,12 +40,34 @@ func main() {
 	storeFor := func(ctx context.Context) (Store, error) { return PgStoreFromEnv(ctx) }
 	b := &Bundle{Env: envFromOS(), Store: storeFor}
 	go hearLoop(storeFor, stdio.Subscribe, time.Sleep)
+	go batchLoop(storeFor)
 	if err := stdio.ServeOn("", b.Tools(), os.Stdin, os.Stdout); err != nil {
 		logf("[mesh-mail-reader] %v", err)
 		os.Exit(1)
 	}
 }
 
+// ---- batches (1427, 1520) --------------------------------------------------------------------------------------
+
+// bundleAsk is the Ask (batch.go) over the runtime's own stdio.Ask.
+func bundleAsk(key string, body map[string]any) (json.RawMessage, error) { return stdio.Ask(key, body) }
+
+// batchLoop runs the reader's batches (RunBatches, batch.go) against the module's own store, judging on this
+// machine (MESH_NODE, the lead's decision at this milestone's prepare step). Started before stdio.ServeOn,
+// exactly as hearLoop is: a batch failure must never take the six mail-reading verbs or mail_reader_health
+// down with it (RunBatches logs and keeps going; see its own doc comment).
+func batchLoop(storeFor func(context.Context) (Store, error)) {
+	store, err := storeFor(context.Background())
+	if err != nil {
+		logf("[mesh-mail-reader] batches: the index cannot be reached: %v", err)
+		return
+	}
+	node := os.Getenv("MESH_NODE")
+	readText := NewReadText(bundleAsk)
+	judge := NewJudge(bundleAsk, node, time.Now)
+	RunBatches(context.Background(), store, readText, judge, time.Now, time.Sleep, logf)
+}
+
 // ---- hearing --------------------------------------------------------------------------------------------------
 
 // hearLoop subscribes to mesh-mail's added events and hears each one (hear.go), exactly as

modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0001-entries-groups-lessons-waiting-mail-and-health-keep-no-body.sql

@@ -4,7 +4,7 @@
 -- taken from one.
 create table mail_group (
     id      bigserial   primary key,
-    label   text        not null default '',
+    label   text        not null unique default '',
     created timestamptz not null default now());
 
 create table mail_entry (
@@ -21,16 +21,27 @@ create table mail_entry (
     mark           text        check (mark is null or mark in ('matters', 'scam warning', 'neither')),
     why            text        not null default '',
     due_date       date,
-    told           boolean     not null default false);
+    told           boolean     not null default false,
+    judged         timestamptz not null default now());
 create index mail_entry_account_date on mail_entry (account, date desc);
+create index mail_entry_judged on mail_entry (judged);
 
+-- A reading lesson (novox/hq ADR 0459 §8, §9, issue 1428): the corrected mail's own headers (never a body),
+-- the change made, when, and whether it was undone. sender_address and mail_group are its own matching keys
+-- for the next batch's question (1427): a lesson matches a mail of the same sender, or already known to be
+-- of the same mail group.
 create table reading_lesson (
-    id      bigserial   primary key,
-    account text        not null default '',
-    headers jsonb       not null default '{}',
-    change  text        not null default '',
-    made    timestamptz not null default now(),
-    undone  boolean     not null default false);
+    id             bigserial   primary key,
+    account        text        not null default '',
+    folder         text        not null default '',
+    sender_name    text        not null default '',
+    sender_address text        not null default '',
+    subject        text        not null default '',
+    mail_group     bigint      references mail_group (id),
+    change         text        not null default '',
+    made           timestamptz not null default now(),
+    undone         boolean     not null default false);
+create index reading_lesson_sender on reading_lesson (sender_address);
 
 -- Heard but not yet read (novox/hq ADR 0459 §3, §12, issue 1518): headers only, kept until reading takes them,
 -- in the order they were heard (id, a bigserial, orders arrival regardless of the mail's own date). A mail

modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore.go

@@ -334,3 +334,185 @@ func (s *PgStore) SetReaderHealth(ctx context.Context, h ReaderHealth) error {
 		h.Healthy, string(h.Cause), h.Why, since)
 	return err
 }
+
+// ---- entries, groups and reading lessons (1427, 1520, 1521, 1428) ----------------------------------------------
+
+func nullDate(t time.Time) *time.Time {
+	if t.IsZero() {
+		return nil
+	}
+	return &t
+}
+
+func (s *PgStore) AddEntry(ctx context.Context, e Entry) (int64, error) {
+	var group *int64
+	if e.Group != 0 {
+		group = &e.Group
+	}
+	var mark *string
+	if e.Mark != "" {
+		m := string(e.Mark)
+		mark = &m
+	}
+	var id int64
+	err := s.pool.QueryRow(ctx, `insert into mail_entry
+		(account, message_id, folder, date, sender_name, sender_address, subject, mail_group, summary, mark, why, due_date, told, judged)
+		values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) returning id`,
+		e.Account, e.MessageID, e.Folder, e.Date, e.SenderName, e.SenderAddress, e.Subject, group, e.Summary, mark, e.Why,
+		nullDate(e.DueDate), e.Told, e.Judged).Scan(&id)
+	return id, err
+}
+
+func (s *PgStore) UpdateEntry(ctx context.Context, e Entry) error {
+	var group *int64
+	if e.Group != 0 {
+		group = &e.Group
+	}
+	var mark *string
+	if e.Mark != "" {
+		m := string(e.Mark)
+		mark = &m
+	}
+	tag, err := s.pool.Exec(ctx, `update mail_entry set account = $2, message_id = $3, folder = $4, date = $5,
+		sender_name = $6, sender_address = $7, subject = $8, mail_group = $9, summary = $10, mark = $11, why = $12,
+		due_date = $13, told = $14, judged = $15 where id = $1`,
+		e.ID, e.Account, e.MessageID, e.Folder, e.Date, e.SenderName, e.SenderAddress, e.Subject, group, e.Summary,
+		mark, e.Why, nullDate(e.DueDate), e.Told, e.Judged)
+	if err != nil {
+		return err
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrNotFound
+	}
+	return nil
+}
+
+func scanEntry(row pgx.Row) (Entry, error) {
+	var e Entry
+	var group *int64
+	var mark *string
+	var dueDate *time.Time
+	err := row.Scan(&e.ID, &e.Account, &e.MessageID, &e.Folder, &e.Date, &e.SenderName, &e.SenderAddress, &e.Subject,
+		&group, &e.Summary, &mark, &e.Why, &dueDate, &e.Told, &e.Judged)
+	if err != nil {
+		return e, err
+	}
+	if group != nil {
+		e.Group = *group
+	}
+	if mark != nil {
+		e.Mark = Mark(*mark)
+	}
+	if dueDate != nil {
+		e.DueDate = *dueDate
+	}
+	e.Date, e.Judged = e.Date.UTC(), e.Judged.UTC()
+	return e, nil
+}
+
+const entryColumns = `id, account, message_id, folder, date, sender_name, sender_address, subject, mail_group, summary, mark, why, due_date, told, judged`
+
+func (s *PgStore) Entry(ctx context.Context, id int64) (Entry, bool, error) {
+	e, err := scanEntry(s.pool.QueryRow(ctx, `select `+entryColumns+` from mail_entry where id = $1`, id))
+	if errors.Is(err, pgx.ErrNoRows) {
+		return Entry{}, false, nil
+	}
+	return e, err == nil, err
+}
+
+func (s *PgStore) Entries(ctx context.Context) ([]Entry, error) {
+	rows, err := s.pool.Query(ctx, `select `+entryColumns+` from mail_entry`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []Entry
+	for rows.Next() {
+		e, err := scanEntry(rows)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, e)
+	}
+	return out, rows.Err()
+}
+
+func (s *PgStore) FindOrCreateGroup(ctx context.Context, label string) (Group, error) {
+	var g Group
+	err := s.pool.QueryRow(ctx, `insert into mail_group (label) values ($1)
+		on conflict (label) do nothing returning id, label, created`, label).Scan(&g.ID, &g.Label, &g.Created)
+	if errors.Is(err, pgx.ErrNoRows) {
+		err = s.pool.QueryRow(ctx, `select id, label, created from mail_group where label = $1`, label).Scan(&g.ID, &g.Label, &g.Created)
+	}
+	if err != nil {
+		return Group{}, err
+	}
+	g.Created = g.Created.UTC()
+	return g, nil
+}
+
+func (s *PgStore) Groups(ctx context.Context) ([]Group, error) {
+	rows, err := s.pool.Query(ctx, `select g.id, g.label, g.created, count(e.id) from mail_group g
+		left join mail_entry e on e.mail_group = g.id group by g.id, g.label, g.created`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []Group
+	for rows.Next() {
+		var g Group
+		if err := rows.Scan(&g.ID, &g.Label, &g.Created, &g.Count); err != nil {
+			return nil, err
+		}
+		g.Created = g.Created.UTC()
+		out = append(out, g)
+	}
+	return out, rows.Err()
+}
+
+func (s *PgStore) AddLesson(ctx context.Context, l Lesson) (int64, error) {
+	var group *int64
+	if l.Group != 0 {
+		group = &l.Group
+	}
+	var id int64
+	err := s.pool.QueryRow(ctx, `insert into reading_lesson
+		(account, folder, sender_name, sender_address, subject, mail_group, change, made, undone)
+		values ($1, $2, $3, $4, $5, $6, $7, $8, $9) returning id`,
+		l.Account, l.Folder, l.SenderName, l.SenderAddress, l.Subject, group, l.Change, l.Made, l.Undone).Scan(&id)
+	return id, err
+}
+
+func (s *PgStore) Lessons(ctx context.Context) ([]Lesson, error) {
+	rows, err := s.pool.Query(ctx, `select id, account, folder, sender_name, sender_address, subject, mail_group, change, made, undone
+		from reading_lesson`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []Lesson
+	for rows.Next() {
+		var l Lesson
+		var group *int64
+		if err := rows.Scan(&l.ID, &l.Account, &l.Folder, &l.SenderName, &l.SenderAddress, &l.Subject, &group, &l.Change, &l.Made, &l.Undone); err != nil {
+			return nil, err
+		}
+		if group != nil {
+			l.Group = *group
+		}
+		l.Made = l.Made.UTC()
+		out = append(out, l)
+	}
+	return out, rows.Err()
+}
+
+func (s *PgStore) UndoLesson(ctx context.Context, id int64) error {
+	tag, err := s.pool.Exec(ctx, `update reading_lesson set undone = true where id = $1`, id)
+	if err != nil {
+		return err
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrNotFound
+	}
+	return nil
+}

modules/mesh-mail-reader/cmd/mesh-mail-reader/schema.sql

@@ -2,7 +2,7 @@
 -- numbered migration for the running store, and this for a fresh install).
 create table mail_group (
     id      bigserial   primary key,
-    label   text        not null default '',
+    label   text        not null unique default '',
     created timestamptz not null default now());
 
 create table mail_entry (
@@ -19,16 +19,27 @@ create table mail_entry (
     mark           text        check (mark is null or mark in ('matters', 'scam warning', 'neither')),
     why            text        not null default '',
     due_date       date,
-    told           boolean     not null default false);
+    told           boolean     not null default false,
+    judged         timestamptz not null default now());
 create index mail_entry_account_date on mail_entry (account, date desc);
+create index mail_entry_judged on mail_entry (judged);
 
+-- A reading lesson (novox/hq ADR 0459 §8, §9, issue 1428): the corrected mail's own headers (never a body),
+-- the change made, when, and whether it was undone. sender_address and mail_group are its own matching keys
+-- for the next batch's question (1427): a lesson matches a mail of the same sender, or already known to be
+-- of the same mail group.
 create table reading_lesson (
-    id      bigserial   primary key,
-    account text        not null default '',
-    headers jsonb       not null default '{}',
-    change  text        not null default '',
-    made    timestamptz not null default now(),
-    undone  boolean     not null default false);
+    id             bigserial   primary key,
+    account        text        not null default '',
+    folder         text        not null default '',
+    sender_name    text        not null default '',
+    sender_address text        not null default '',
+    subject        text        not null default '',
+    mail_group     bigint      references mail_group (id),
+    change         text        not null default '',
+    made           timestamptz not null default now(),
+    undone         boolean     not null default false);
+create index reading_lesson_sender on reading_lesson (sender_address);
 
 -- Heard but not yet read (novox/hq ADR 0459 §3, §12, issue 1518): headers only, kept until reading takes them,
 -- in the order they were heard (id, a bigserial, orders arrival regardless of the mail's own date). A mail

modules/mesh-mail-reader/cmd/mesh-mail-reader/store.go

@@ -41,13 +41,18 @@ type Entry struct {
 	Why           string
 	DueDate       time.Time // zero when none
 	Told          bool
+	// Judged is when a batch judged this mail (issue 1520) — the week the weekly measure (1521/AC3) counts it
+	// in, which is never the mail's own Date: a mail that arrived long before it was judged is read this week.
+	Judged time.Time
 }
 
-// Group is one mail group (ADR 0459 §4).
+// Group is one mail group (ADR 0459 §4). Count, how many entries hold it, is answered by Store.Groups
+// (1521/AC2) and is never itself kept.
 type Group struct {
 	ID      int64
 	Label   string
 	Created time.Time
+	Count   int
 }
 
 // Lesson is one reading lesson (ADR 0459 §8, §9): the corrected mail's own headers (sender name and address,
@@ -56,8 +61,7 @@ type Group struct {
 // mail of the batch"); Group, when not zero, is a second matching key for a mail already known to be of that
 // group (the same clause's "or mail group") — no issue yet looks up a mail's group before it is judged, so in
 // practice only SenderAddress matching fires today (1427's own tests); the field and the comparison both exist
-// so that lookup can be added later without reshaping this type. The store's own persistence of lessons
-// (adding one, listing them, undoing one) is 1428's, not this type's.
+// so that lookup can be added later without reshaping this type.
 type Lesson struct {
 	ID            int64
 	Account       string
@@ -130,11 +134,42 @@ type Store interface {
 	AddWaiting(ctx context.Context, w WaitingMail) error
 	// Waiting answers every waiting mail in arrival order, oldest first (1518/AC4).
 	Waiting(ctx context.Context) ([]WaitingMail, error)
-	// TakeWaiting removes a waiting mail once it has been handed to reading (1520); not yet called before 1519.
+	// TakeWaiting removes a waiting mail once it has been handed to reading (1520).
 	TakeWaiting(ctx context.Context, seq int64) error
 
 	// ReaderHealth answers the kept health state (1519/AC4); healthy with a zero Since on a fresh database.
 	ReaderHealth(ctx context.Context) (ReaderHealth, error)
 	// SetReaderHealth replaces the kept health state.
 	SetReaderHealth(ctx context.Context, h ReaderHealth) error
+
+	// AddEntry records a new mail entry (1520/AC3) and answers its own id.
+	AddEntry(ctx context.Context, e Entry) (int64, error)
+	// UpdateEntry replaces an entry's fields by its id (1428/AC1, correct). ErrNotFound when unknown.
+	UpdateEntry(ctx context.Context, e Entry) error
+	// Entry answers one entry by id, and whether it was found (1428/AC1, AC5: an unknown id is refused).
+	Entry(ctx context.Context, id int64) (Entry, bool, error)
+	// Entries answers every mail entry; ordering and filtering for the verb are 1521's own pure functions, not
+	// the store's.
+	Entries(ctx context.Context) ([]Entry, error)
+
+	// FindOrCreateGroup answers the mail group of this label, making one if none yet carries it (1520/AC3): the
+	// judge names groups by their label, never by id.
+	FindOrCreateGroup(ctx context.Context, label string) (Group, error)
+	// Groups answers every mail group, each with how many entries hold it (1521/AC2).
+	Groups(ctx context.Context) ([]Group, error)
+
+	// AddLesson keeps one reading lesson at once (1428/AC1) and answers its own id.
+	AddLesson(ctx context.Context, l Lesson) (int64, error)
+	// Lessons answers every reading lesson, undone ones included (1427's own selection, and 1428's lessons
+	// verb, each decide for themselves what to do with Undone).
+	Lessons(ctx context.Context) ([]Lesson, error)
+	// UndoLesson marks one reading lesson undone (1428/AC3). ErrNotFound when unknown.
+	UndoLesson(ctx context.Context, id int64) error
 }
+
+// ErrNotFound is answered by a Store method asked for something it does not hold.
+var ErrNotFound = notFoundError{}
+
+type notFoundError struct{}
+
+func (notFoundError) Error() string { return "not found" }