From 42bffb93636b5dceec65c85d2885b22b6fea4809 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 00:18:37 +0200 Subject: [PATCH] Run the reader's batches: paced, judged, written as entries, unhealthy on failure (issue 1520) --- .../cmd/mesh-mail-reader/batch.go | 252 +++++++++++++++ .../cmd/mesh-mail-reader/batch_test.go | 305 ++++++++++++++++++ .../cmd/mesh-mail-reader/kit_test.go | 107 ++++++ .../cmd/mesh-mail-reader/main.go | 34 +- ...s-waiting-mail-and-health-keep-no-body.sql | 27 +- .../cmd/mesh-mail-reader/pgstore.go | 182 +++++++++++ .../cmd/mesh-mail-reader/schema.sql | 27 +- .../cmd/mesh-mail-reader/store.go | 43 ++- 8 files changed, 952 insertions(+), 25 deletions(-) create mode 100644 modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go create mode 100644 modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_test.go diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go new file mode 100644 index 00000000..db22c949 --- /dev/null +++ b/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) + } + } +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_test.go new file mode 100644 index 00000000..74b2ce32 --- /dev/null +++ b/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) + } + } + } +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/kit_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/kit_test.go index abb021ee..6d831299 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/kit_test.go +++ b/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)") diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go index b323b196..3a16b424 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go +++ b/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 diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0001-entries-groups-lessons-waiting-mail-and-health-keep-no-body.sql b/modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0001-entries-groups-lessons-waiting-mail-and-health-keep-no-body.sql index f3712f14..e443629c 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0001-entries-groups-lessons-waiting-mail-and-health-keep-no-body.sql +++ b/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 diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore.go index 92289b16..ad374f91 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore.go +++ b/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 +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema.sql b/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema.sql index c3383487..9c592cb9 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema.sql +++ b/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 diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/store.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/store.go index e018cb4b..a93c6575 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/store.go +++ b/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" }