issue 1857: read the last 14 days of inbox mail once more (1857.5)
ADR 0516 §5: so the operator's list does not start empty, the inbox mail of the last 14 days of each granted account is read again in ordinary batches, after any mail just heard; it is kept as done in its own table (migration 0004), and the journal says its start, end and count. The reader now invokes mail-store.search.
This commit is contained in:
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
}
|
||||
|
||||
+8
@@ -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));
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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: "<old@example.org>", 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)
|
||||
}
|
||||
}
|
||||
@@ -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:
|
||||
|
||||
@@ -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));
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user