diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go index bd40d673..c66866b3 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch.go @@ -203,6 +203,14 @@ func recordOutcome(ctx context.Context, store Store, outcome HealthOutcome, why // (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, logf func(string, ...any)) error { + take := func(w WaitingMail, _ bool) error { return store.TakeWaiting(ctx, w.Seq) } + return runBatch(ctx, store, readText, judge, waiting, now, logf, take) +} + +// runBatch is RunOneBatch with what becomes of a mail that is done — judged, or gone — given as take: a +// waiting mail leaves the waiting queue (RunOneBatch); a re-read one leaves the re-reading (rereading.go). +func runBatch(ctx context.Context, store Store, readText ReadTextFunc, judge JudgeFunc, waiting []WaitingMail, now func() time.Time, + logf func(string, ...any), take func(w WaitingMail, judged bool) error) error { if len(waiting) > MaxMailsPerBatch { waiting = waiting[:MaxMailsPerBatch] } @@ -225,7 +233,7 @@ func RunOneBatch(ctx context.Context, store Store, readText ReadTextFunc, judge text, err := readText(w.Account, w.Folder, w.UID) if err != nil { if isNoMessageError(err) { - if takeErr := store.TakeWaiting(ctx, w.Seq); takeErr != nil { + if takeErr := take(w, false); takeErr != nil { return takeErr } logf("[mesh-mail-reader] %s %s uid %d is gone (moved or deleted since it was heard): taken out of the waiting queue, never judged", w.Account, w.Folder, w.UID) @@ -282,15 +290,26 @@ func RunOneBatch(ctx context.Context, store Store, readText ReadTextFunc, judge if _, err := store.AddEntry(ctx, e); err != nil { return err } - if err := store.TakeWaiting(ctx, w.Seq); err != nil { + if err := take(w, true); err != nil { return err } } return recordOutcome(ctx, store, OutcomeOK, "", judgedAt) } +// finishRereading keeps that the re-reading was done (ADR 0516 §5) and answers nil, or keeps it running when +// that could not be kept, so the next turn tries again. +func finishRereading(ctx context.Context, store Store, r *Rereading, now time.Time, logf func(string, ...any)) *Rereading { + if err := r.finish(ctx, store, now, logf); err != nil { + logf("[mesh-mail-reader] the 14-day re-reading's end could not be kept: %v", err) + return r + } + return nil +} + // RunBatches is the loop the running module drives (main.go): pace batch starts at least gap apart -// (nextBatchWait), take at most MaxMailsPerBatch of the waiting mails in arrival order, run one batch at a +// (nextBatchWait), take at most MaxMailsPerBatch of the waiting mails in arrival order — or, while none +// waits, of the 14-day re-reading's mails (reread, nil when there is none; rereading.go) — run one batch at a // time, and after each one — win or lose — try the mail signal (SendSignal, tell.go, issue 1522; it is its // own no-op when there is nothing untold that matters, or the reader is unhealthy) — forever, until ctx is // done. gap is given explicitly (main.go's batchLoop passes the var BatchGap, which only a `-tags @@ -300,7 +319,7 @@ func RunOneBatch(ctx context.Context, store Store, readText ReadTextFunc, judge // when RunBatches read BatchGap directly inside its own loop (review round 2, issue 1520). 1520/AC2-AC5's // one-batch behaviour and 1522's signal-sending are proven directly, on RunOneBatch and SendSignal, in // batch_test.go and tell_test.go. -func RunBatches(ctx context.Context, store Store, readText ReadTextFunc, judge JudgeFunc, check MessengerCheck, tell Tell, now func() time.Time, sleep func(time.Duration), logf func(string, ...any), gap time.Duration) { +func RunBatches(ctx context.Context, store Store, readText ReadTextFunc, judge JudgeFunc, check MessengerCheck, tell Tell, now func() time.Time, sleep func(time.Duration), logf func(string, ...any), gap time.Duration, reread *Rereading) { var lastStart time.Time for { if ctx.Err() != nil { @@ -316,17 +335,31 @@ func RunBatches(ctx context.Context, store Store, readText ReadTextFunc, judge J sleep(time.Second) continue } - if len(waiting) == 0 { + switch { + case len(waiting) > 0: + // A mail just heard goes first, before the re-reading's next batch (1857/AC6). + if len(waiting) > MaxMailsPerBatch { + waiting = waiting[:MaxMailsPerBatch] + } + lastStart = now() + if err := RunOneBatch(ctx, store, readText, judge, waiting, now, logf); err != nil { + logf("[mesh-mail-reader] a batch failed: %v", err) + } + case reread != nil && len(reread.Pending) > 0: + lastStart = now() + if err := runBatch(ctx, store, readText, judge, reread.next(), now, logf, reread.take); err != nil { + logf("[mesh-mail-reader] a batch of the re-reading failed: %v", err) + } + if len(reread.Pending) == 0 { + reread = finishRereading(ctx, store, reread, now(), logf) + } + default: + if reread != nil { // a re-reading that found no mail at all + reread = finishRereading(ctx, store, reread, now(), logf) + } sleep(time.Second) continue } - if len(waiting) > MaxMailsPerBatch { - waiting = waiting[:MaxMailsPerBatch] - } - lastStart = now() - if err := RunOneBatch(ctx, store, readText, judge, waiting, now, logf); err != nil { - logf("[mesh-mail-reader] a batch failed: %v", err) - } if err := SendSignal(ctx, store, check, tell); err != nil { logf("[mesh-mail-reader] the mail signal could not be sent: %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 index 21bfe180..7436087d 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_test.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/batch_test.go @@ -122,7 +122,7 @@ func TestThirtyWaitingMailsRunAsFiveBatchesAtLeast160sApart(t *testing.T) { done := make(chan struct{}) go func() { - RunBatches(ctx, store, noopReadText, judge, check, tell, now, sleep, noopLogf, MinBatchGap) + RunBatches(ctx, store, noopReadText, judge, check, tell, now, sleep, noopLogf, MinBatchGap, nil) close(done) }() select { 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 84a1dfbe..d1918346 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 @@ -8,6 +8,7 @@ import ( "context" "errors" "sync" + "time" ) type memStore struct { @@ -23,6 +24,26 @@ type memStore struct { nextGroup int64 lessons []Lesson nextLesson int64 + + reread *rereadKept +} + +type rereadKept struct { + started, ended time.Time + mails int +} + +func (s *memStore) RereadingDone(context.Context) (bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.reread != nil, nil +} + +func (s *memStore) KeepRereading(_ context.Context, started, ended time.Time, mails int) error { + s.mu.Lock() + defer s.mu.Unlock() + s.reread = &rereadKept{started, ended, mails} + return nil } func newMemStore() *memStore { 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 e85b95a2..2104558e 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/main.go @@ -100,9 +100,32 @@ func batchLoop(storeFor func(context.Context) (Store, error)) { judge := NewJudge(bundleAsk, node, time.Now) check := NewMessengerCheck(bundleAsk) tell := NewTell(bundleAsk) - RunBatches(context.Background(), store, readText, judge, check, tell, time.Now, time.Sleep, logf, BatchGap) + RunBatches(context.Background(), store, readText, judge, check, tell, time.Now, time.Sleep, logf, BatchGap, startRereading(store, time.Sleep)) } +// startRereading starts the 14-day re-reading (ADR 0516 §5) unless it was done, asking mail-store up to +// rereadTries times rereadRetry apart while it does not answer (mail-store may still be starting); after that +// the batches start without it, the journal says so, and the next start of the module tries again. +func startRereading(store Store, sleep func(time.Duration)) *Rereading { + for try := 1; ; try++ { + r, err := StartRereading(context.Background(), store, bundleAsk, time.Now, logf) + if err == nil { + return r + } + if try == rereadTries { + logf("[mesh-mail-reader] the 14-day re-reading cannot start (%v): it waits for the module's next start", err) + return nil + } + logf("[mesh-mail-reader] the 14-day re-reading cannot start yet (%v); trying again in %s", err, rereadRetry) + sleep(rereadRetry) + } +} + +const ( + rereadTries = 5 + rereadRetry = 30 * time.Second +) + // ---- hearing -------------------------------------------------------------------------------------------------- // hearFirstRetryWait is the hearing loop's first retry wait (issue 1518, fixed per the lead's try feedback diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/manifest_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/manifest_test.go index b4d1a4e7..5f1723e8 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/manifest_test.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/manifest_test.go @@ -64,10 +64,11 @@ func TestTheManifestClaimsMailReadingAndInvokesReadOnly(t *testing.T) { if !reflect.DeepEqual(m.Claims, wantClaims) { t.Fatalf("claims %+v, want %+v", m.Claims, wantClaims) } - // mail-store.read and mail-store.accounts only (1426/AC2); claude_code_judge/judged (1520), messenger_check - // and operator-channel.tell (1522) are its other invokes, added as each landed — never send or send-status. + // mail-store.read and mail-store.accounts only (1426/AC2), and mail-store.search for the 14-day re-reading + // (1857/AC6); claude_code_judge/judged (1520), messenger_check and operator-channel.tell (1522) are its other + // invokes, added as each landed — never send or send-status. wantInvokes := []string{ - "seat:mail-store.read", "seat:mail-store.accounts", + "seat:mail-store.read", "seat:mail-store.accounts", "seat:mail-store.search", "claude-code.claude_code_judge", "claude-code.claude_code_judged", "messenger.messenger_check", "seat:operator-channel.tell", } diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0004-tasks-promises-and-read-later.sql b/modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0004-tasks-promises-and-read-later.sql index fbeec69d..33c0ea39 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0004-tasks-promises-and-read-later.sql +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/migrations/0004-tasks-promises-and-read-later.sql @@ -16,3 +16,11 @@ create table mail_promise ( who text not null default '' check (char_length(who) <= 120), date date, primary key (mail_entry, n)); + +-- The 14-day re-reading (ADR 0516 §5): once done, kept here, so a restart never starts it again. +create table rereading ( + id smallint primary key default 1, + started timestamptz not null, + ended timestamptz not null, + mails integer not null, + constraint rereading_once check (id = 1)); 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 9a35066b..27af7f75 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/pgstore.go @@ -636,6 +636,19 @@ func (s *PgStore) Lessons(ctx context.Context) ([]Lesson, error) { return out, rows.Err() } +func (s *PgStore) RereadingDone(ctx context.Context) (bool, error) { + var done bool + err := s.pool.QueryRow(ctx, `select exists (select 1 from rereading)`).Scan(&done) + return done, err +} + +func (s *PgStore) KeepRereading(ctx context.Context, started, ended time.Time, mails int) error { + _, err := s.pool.Exec(ctx, `insert into rereading (id, started, ended, mails) values (1, $1, $2, $3) + on conflict (id) do update set started = excluded.started, ended = excluded.ended, mails = excluded.mails`, + started, ended, mails) + return 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 { diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/rereading.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/rereading.go new file mode 100644 index 00000000..5c92a66f --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/rereading.go @@ -0,0 +1,155 @@ +package main + +// The 14-day re-reading (novox/hq ADR 0516 §5, issue 1857/AC6): once this is delivered, mail reading reads the +// inbox mails of the last 14 days of each granted account once more, so the operator's list starts full. It +// runs in ordinary batches under the batch loop's own spacing (ADR 0459 §15) — a mail just heard goes first — +// keeps that it was done, and says its start, its end and how many mails it read in the journal. A mail +// already judged keeps its entry (and its id); its judgement is replaced with one that holds its tasks, +// promises and read-later (1849's one entry per message). + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "time" +) + +// RereadDays is how far back the re-reading reads. +const RereadDays = 14 + +// rereadPage is how many mails one search asks: mail-store.search's own most (cmd/mesh-mail/tools.go +// MostResults). +const rereadPage = 50 + +// rereadFolder is the inbox, by the name mesh-mail gives it for every kind of account. +const rereadFolder = "INBOX" + +// Rereading is a re-reading under way: the mails still to read, when it started and how many it has read. +type Rereading struct { + Pending []WaitingMail + Started time.Time + Read int +} + +// StartRereading answers the re-reading to run, or nil when it was done already (a second start reads +// nothing). It asks mail-store which accounts are granted, and each one's inbox mails dated within the last +// RereadDays days, newest first, a page at a time; the mails are read oldest first. Nothing is kept until it +// ends (RunBatches): a restart before then starts it again. +func StartRereading(ctx context.Context, store Store, ask Ask, now func() time.Time, logf func(string, ...any)) (*Rereading, error) { + done, err := store.RereadingDone(ctx) + if err != nil || done { + return nil, err + } + started := now() + cutoff := started.Add(-RereadDays * 24 * time.Hour) + raw, err := ask("seat:mail-store.accounts", map[string]any{}) + if err != nil { + return nil, err + } + var granted struct { + Accounts []struct { + Account string `json:"account"` + } `json:"accounts"` + } + if err := json.Unmarshal(raw, &granted); err != nil { + return nil, fmt.Errorf("mail-store.accounts' answer is not a list of accounts: %.200s", raw) + } + r := &Rereading{Started: started} + for _, a := range granted.Accounts { + mails, err := inboxSince(ask, a.Account, cutoff) + if err != nil { + return nil, err + } + r.Pending = append(r.Pending, mails...) + } + sort.SliceStable(r.Pending, func(i, j int) bool { return r.Pending[i].Date.Before(r.Pending[j].Date) }) + for i := range r.Pending { + r.Pending[i].Seq = int64(i + 1) // the question's own id for it; never a waiting mail's + } + logf("[mesh-mail-reader] the 14-day re-reading starts: %d inbox mails since %s in %d accounts", + len(r.Pending), cutoff.UTC().Format(time.RFC3339), len(granted.Accounts)) + return r, nil +} + +// inboxSince is one account's inbox mails dated on or after cutoff, a page of newest first at a time, each +// next page before the oldest one seen. +func inboxSince(ask Ask, account string, cutoff time.Time) ([]WaitingMail, error) { + var out []WaitingMail + seen := map[string]bool{} + before := "" + for { + args := map[string]any{"account": account, "folder": rereadFolder, "on-or-after": cutoff.UTC().Format(time.RFC3339), + "order-by": "date", "order": "desc", "limit": rereadPage} + if before != "" { + args["before"] = before + } + raw, err := ask("seat:mail-store.search", args) + if err != nil { + return nil, err + } + var page struct { + Found []EventBody `json:"found"` + } + if err := json.Unmarshal(raw, &page); err != nil { + return nil, fmt.Errorf("mail-store.search's answer is not a list of mails: %.200s", raw) + } + added := 0 + var oldest time.Time + for _, m := range page.Found { + date, err := time.Parse(time.RFC3339, m.Date) + if err != nil || date.Before(cutoff) || m.Folder != rereadFolder { + continue + } + w := WaitingMail{Account: m.Account, Folder: m.Folder, MessageID: m.MessageID, UIDValidity: m.UIDValidity, + UID: m.UID, Date: date, From: m.From, Subject: m.Subject, Flags: m.Flags, Size: m.Size} + if seen[w.Key()] { + continue + } + seen[w.Key()] = true + out = append(out, w) + added++ + if oldest.IsZero() || date.Before(oldest) { + oldest = date + } + } + // The last page, or one that brought nothing new (every mail on it older than cutoff, or already + // seen at the same second): this account is done. + if len(page.Found) < rereadPage || added == 0 { + return out, nil + } + // before is exclusive: a second past the oldest, so mails of that same second are asked again (and + // passed over as seen) rather than lost. + before = oldest.Add(time.Second).UTC().Format(time.RFC3339) + } +} + +// take is the re-reading's own take (runBatch): the mail leaves Pending, and is counted when it was judged. +func (r *Rereading) take(w WaitingMail, judged bool) error { + for i, p := range r.Pending { + if p.Seq == w.Seq { + r.Pending = append(r.Pending[:i], r.Pending[i+1:]...) + break + } + } + if judged { + r.Read++ + } + return nil +} + +// next is the re-reading's next batch: at most MaxMailsPerBatch of what is still pending, in order. +func (r *Rereading) next() []WaitingMail { + n := min(len(r.Pending), MaxMailsPerBatch) + return append([]WaitingMail(nil), r.Pending[:n]...) +} + +// finish keeps that the re-reading was done and says so in the journal. +func (r *Rereading) finish(ctx context.Context, store Store, now time.Time, logf func(string, ...any)) error { + if err := store.KeepRereading(ctx, r.Started, now, r.Read); err != nil { + return err + } + logf("[mesh-mail-reader] the 14-day re-reading ended: %d mails read, started %s, ended %s", + r.Read, r.Started.UTC().Format(time.RFC3339), now.UTC().Format(time.RFC3339)) + return nil +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/rereading_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/rereading_test.go new file mode 100644 index 00000000..3f489921 --- /dev/null +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/rereading_test.go @@ -0,0 +1,226 @@ +package main + +// Issue 1857/AC6 (novox/hq ADR 0516 §5): the inbox mail of the last 14 days of each granted account is read +// once more, in ordinary batches 160 s apart; it is kept as done, and the journal says its start, end and +// count. + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "sync" + "testing" + "time" +) + +// fakeMailStore answers seat:mail-store.accounts and seat:mail-store.search in mesh-mail's own shapes +// (accountsTool and searchTool, cmd/mesh-mail/tools.go; metadata, cmd/mesh-mail/imap.go): newest first, at +// most limit (50 at most), within account, folder and before. It does NOT apply on-or-after, so the reader's +// own 14-day bound is what this test proves; it records the on-or-after it was asked. +type fakeMailStore struct { + mails []map[string]any + searched int + onOrAfter []string +} + +func newFakeMailStore(now time.Time, days int) *fakeMailStore { + f := &fakeMailStore{} + uid := 0 + for _, account := range []string{"work", "personal"} { + for d := 0; d < days; d++ { + for h := 1; h <= 3; h++ { + uid++ + for _, folder := range []string{"INBOX", "Sent"} { + f.mails = append(f.mails, map[string]any{ + "account": account, "folder": folder, "uid": uid, "uidvalidity": 1, + "message-id": fmt.Sprintf("<%s-%d-%s@example.org>", account, uid, folder), "in-reply-to": "", + "date": now.Add(-time.Duration(d)*24*time.Hour - time.Duration(h)*time.Hour).UTC().Format(time.RFC3339), + "from": []map[string]any{{"name": "Sara Claes", "address": "sara@harbor.example"}}, + "to": []map[string]any{{"address": "op@example.org"}}, "cc": []any{}, + "subject": fmt.Sprintf("mail %d", uid), "size": 1000, "flags": []string{}, "attachments": []any{}, + }) + } + } + } + } + return f +} + +func (f *fakeMailStore) ask(key string, body map[string]any) (json.RawMessage, error) { + switch key { + case "seat:mail-store.accounts": + return json.Marshal(map[string]any{"accounts": []map[string]any{ + {"account": "work", "kind": "imap", "address": "op@work.example", "state": "synced", "folders": []string{"INBOX"}}, + {"account": "personal", "kind": "gmail", "address": "op@mail.example", "state": "synced", "folders": []string{"INBOX"}}, + }, "caller": "mesh-mail-reader"}) + case "seat:mail-store.search": + f.searched++ + after, _ := body["on-or-after"].(string) + f.onOrAfter = append(f.onOrAfter, after) + limit := 20 + if l, ok := body["limit"].(int); ok { + limit = l + } + if limit > 50 { + return nil, fmt.Errorf("limit is 1 to 50") + } + var before time.Time + if s, ok := body["before"].(string); ok && s != "" { + before, _ = time.Parse(time.RFC3339, s) + } + var found []map[string]any + for _, m := range f.mails { + date, _ := time.Parse(time.RFC3339, m["date"].(string)) + if m["account"] != body["account"] || m["folder"] != body["folder"] || (!before.IsZero() && !date.Before(before)) { + continue + } + found = append(found, m) + } + sort.SliceStable(found, func(i, j int) bool { return found[i]["date"].(string) > found[j]["date"].(string) }) + if len(found) > limit { + found = found[:limit] + } + return json.Marshal(map[string]any{"searched": []string{body["account"].(string)}, "found": found}) + } + return nil, fmt.Errorf("the fake mail-store does not answer %s", key) +} + +// 1857/AC6. +func TestTheRereadingReadsTheLast14DaysOnceInSpacedBatches(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + store := newMemStore() + var clockMu sync.Mutex + clock := time.Date(2026, 10, 22, 12, 0, 0, 0, time.UTC) + now := func() time.Time { clockMu.Lock(); defer clockMu.Unlock(); return clock } + started := clock + sleep := func(d time.Duration) { + clockMu.Lock() + clock = clock.Add(d) + if clock.Sub(started) > 48*time.Hour { // a loop that never ends its work ends the test instead + cancel() + } + clockMu.Unlock() + } + mailStore := newFakeMailStore(clock, 20) // 20 days, 3 inbox mails a day, in two granted accounts + var journal []string + var journalMu sync.Mutex + logf := func(format string, a ...any) { + journalMu.Lock() + defer journalMu.Unlock() + journal = append(journal, fmt.Sprintf(format, a...)) + } + + reread, err := StartRereading(ctx, store, mailStore.ask, now, logf) + if err != nil { + t.Fatal(err) + } + want := 2 * 14 * 3 // the last 14 days only + batches := (want + MaxMailsPerBatch - 1) / MaxMailsPerBatch + var starts []time.Time + var subjects []string + judge := func(question string, _ map[string]any) (json.RawMessage, HealthOutcome, string) { + starts = append(starts, now()) + var mails []map[string]any + for _, line := range strings.Split(question, "\n") { + var id int64 + if _, err := fmt.Sscanf(line, "mail %d:", &id); err == nil { + mails = append(mails, map[string]any{"id": id, "mark": "neither"}) + subjects = append(subjects, line) + } + } + if len(starts) == batches { + cancel() + } + raw, _ := json.Marshal(map[string]any{"mails": mails}) + return raw, OutcomeOK, "" + } + check := func(string) (bool, error) { return true, nil } + tell := func(string, []string, string) (string, error) { return "", nil } + done := make(chan struct{}) + go func() { + RunBatches(ctx, store, noopReadText, judge, check, tell, now, sleep, logf, MinBatchGap, reread) + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatalf("RunBatches did not end; %d batches judged", len(starts)) + } + + entries, _ := store.Entries(context.Background()) + if len(entries) != want { + t.Fatalf("%d entries, want %d: the inbox mail of the last 14 days in both accounts, once", len(entries), want) + } + cutoff := started.Add(-RereadDays * 24 * time.Hour) + for _, e := range entries { + if e.Date.Before(cutoff) || e.Folder != "INBOX" { + t.Fatalf("read %s %s of %s, outside the last 14 days' inbox", e.Account, e.Folder, e.Date) + } + } + if len(starts) != batches { + t.Fatalf("%d batches, want %d of at most %d", len(starts), batches, MaxMailsPerBatch) + } + for i := 1; i < len(starts); i++ { + if gap := starts[i].Sub(starts[i-1]); gap < MinBatchGap { + t.Fatalf("batch %d started %s after the one before, want at least %s", i, gap, MinBatchGap) + } + } + for _, asked := range mailStore.onOrAfter { + if asked != cutoff.Format(time.RFC3339) { + t.Fatalf("searched on-or-after %q, want %q", asked, cutoff.Format(time.RFC3339)) + } + } + if store.reread == nil || store.reread.mails != want || !store.reread.started.Equal(started) { + t.Fatalf("kept %+v, want done with %d mails, started %s", store.reread, want, started) + } + all := strings.Join(journal, "\n") + if !strings.Contains(all, "the 14-day re-reading starts") || !strings.Contains(all, "the 14-day re-reading ended") || + !strings.Contains(all, fmt.Sprintf("%d mails read", want)) { + t.Fatalf("the journal does not say the start, the end and the count:\n%s", all) + } + + // A second start reads nothing: not even a search. + searched := mailStore.searched + again, err := StartRereading(context.Background(), store, mailStore.ask, now, logf) + if err != nil || again != nil || mailStore.searched != searched { + t.Fatalf("a second start: %+v, %v, searched %d more; want nothing", again, err, mailStore.searched-searched) + } +} + +// 1857/AC6: a mail just heard is read before the re-reading's next batch. +func TestAMailJustHeardGoesBeforeTheRereading(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + store := newMemStore() + seedWaiting(t, store, 1) + reread := &Rereading{Pending: []WaitingMail{{Seq: 1, Account: "work", Folder: "INBOX", MessageID: "", UID: 99}}} + var order []string + judge := func(question string, _ map[string]any) (json.RawMessage, HealthOutcome, string) { + var id int64 + for _, line := range strings.Split(question, "\n") { + if _, err := fmt.Sscanf(line, "mail %d:", &id); err == nil { + order = append(order, line) + } + } + if len(order) == 2 { + cancel() + } + raw, _ := json.Marshal(map[string]any{"mails": []map[string]any{{"id": id, "mark": "neither"}}}) + return raw, OutcomeOK, "" + } + clock := time.Date(2026, 10, 22, 12, 0, 0, 0, time.UTC) + RunBatches(ctx, store, noopReadText, judge, func(string) (bool, error) { return true, nil }, + func(string, []string, string) (string, error) { return "", nil }, + func() time.Time { return clock }, func(d time.Duration) { + clock = clock.Add(d) + if clock.Hour() == 0 { // a loop that never ends its work ends the test instead + cancel() + } + }, noopLogf, MinBatchGap, reread) + if len(order) != 2 || !strings.Contains(order[0], "account bob") || !strings.Contains(order[1], "account work") { + t.Fatalf("judged %v, want the heard mail (bob) first, then the re-read one (work)", order) + } +} diff --git a/modules/mesh-mail-reader/cmd/mesh-mail-reader/runtime_player_integration_test.go b/modules/mesh-mail-reader/cmd/mesh-mail-reader/runtime_player_integration_test.go index f75322f5..fd580219 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/runtime_player_integration_test.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/runtime_player_integration_test.go @@ -73,6 +73,7 @@ type standIns struct { mu sync.Mutex MailAccounts func(args map[string]any) (any, error) + MailSearch func(args map[string]any) (any, error) MailRead func(args map[string]any) (any, error) Judge func(args map[string]any) (any, error) Judged func(args map[string]any) (any, error) @@ -89,8 +90,16 @@ func (s *standIns) answer(key string, args map[string]any) (any, error, bool) { s.mu.Unlock() var f func(args map[string]any) (any, error) switch { + case key == "seat:mail-store.accounts" && s.MailAccounts == nil, + key == "seat:mail-store.search" && s.MailSearch == nil: + // The 14-day re-reading (issue 1857/AC6) asks these at the bundle's start, before a test has given + // its stand-ins: read-only, so a test that gave none is answered no account and no mail, and the + // re-reading ends at once, reading nothing. + return map[string]any{"accounts": []any{}, "searched": []any{}, "found": []any{}}, nil, true case key == "seat:mail-store.accounts": f = s.MailAccounts + case key == "seat:mail-store.search": + f = s.MailSearch case key == "seat:mail-store.read": f = s.MailRead case key == "claude-code.claude_code_judge@"+testNode: 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 e4955b4e..b91f57a2 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema.sql +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/schema.sql @@ -98,3 +98,11 @@ create table reader_health ( since timestamptz, last_success timestamptz, constraint reader_health_singleton check (id = 1)); + +-- The 14-day re-reading (ADR 0516 §5): once done, kept here, so a restart never starts it again. +create table rereading ( + id smallint primary key default 1, + started timestamptz not null, + ended timestamptz not null, + mails integer not null, + constraint rereading_once check (id = 1)); 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 da097aa1..73bd19e2 100644 --- a/modules/mesh-mail-reader/cmd/mesh-mail-reader/store.go +++ b/modules/mesh-mail-reader/cmd/mesh-mail-reader/store.go @@ -235,6 +235,11 @@ type Store interface { Lessons(ctx context.Context) ([]Lesson, error) // UndoLesson marks one reading lesson undone (1428/AC3). ErrNotFound when unknown. UndoLesson(ctx context.Context, id int64) error + + // RereadingDone says whether the 14-day re-reading (ADR 0516 §5, issue 1857) was done; KeepRereading keeps + // that it was, when it started and ended and how many mails it read, so a restart never starts it again. + RereadingDone(ctx context.Context) (bool, error) + KeepRereading(ctx context.Context, started, ended time.Time, mails int) error } // ErrNotFound is answered by a Store method asked for something it does not hold. diff --git a/modules/mesh-mail-reader/module.json b/modules/mesh-mail-reader/module.json index b83e8bb4..80ecdeea 100644 --- a/modules/mesh-mail-reader/module.json +++ b/modules/mesh-mail-reader/module.json @@ -15,6 +15,7 @@ "invokes": [ "seat:mail-store.read", "seat:mail-store.accounts", + "seat:mail-store.search", "claude-code.claude_code_judge", "claude-code.claude_code_judged", "messenger.messenger_check",