jochen
f3007eb242
mesh/merge-gate head f3007eb242f8 merged into trunk 6c6d5f681c30 — pass: builds mesh-mail, mesh-mail-web, new: modules/mesh-mail-reader → novox; no bus ste…
mesh/repo-check head f3007eb242f8 merged into trunk 6c6d5f681c30 — pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-…
mesh/delivery delivered
443 lines
17 KiB
Go
443 lines
17 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// noopLogf is the logf RunOneBatch and RunBatches want, for a test that does not care what was logged.
|
|
func noopLogf(string, ...any) {}
|
|
|
|
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 MinBatchGap has passed since the last start, or no batch has
|
|
// started yet; the remainder otherwise. Passes MinBatchGap explicitly, never the var BatchGap, so this
|
|
// still proves 160s even when compiled into a binary whose BatchGap has been shortened (batch_gap_integration.go).
|
|
func TestNextBatchWait(t *testing.T) {
|
|
start := time.Date(2026, 10, 20, 10, 0, 0, 0, time.UTC)
|
|
if w := nextBatchWait(time.Time{}, start, MinBatchGap); w != 0 {
|
|
t.Fatalf("no batch yet: wait = %s, want 0", w)
|
|
}
|
|
if w := nextBatchWait(start, start.Add(100*time.Second), MinBatchGap); w != 60*time.Second {
|
|
t.Fatalf("100s after: wait = %s, want 60s", w)
|
|
}
|
|
if w := nextBatchWait(start, start.Add(MinBatchGap), MinBatchGap); w != 0 {
|
|
t.Fatalf("exactly MinBatchGap after: wait = %s, want 0 (one waiting mail starts a batch at once)", w)
|
|
}
|
|
if w := nextBatchWait(start, start.Add(MinBatchGap+time.Second), MinBatchGap); w != 0 {
|
|
t.Fatalf("past MinBatchGap: wait = %s, want 0", w)
|
|
}
|
|
}
|
|
|
|
// 1520/AC1 (review round 1 finding 5): with 30 mails waiting, RunBatches ITSELF — never a hand-driven copy
|
|
// of its own loop — runs three batches of 10, their starts at least MinBatchGap (160s) apart. A test clock:
|
|
// now reads a clock this test owns, and sleep (the only thing RunBatches calls to wait) advances that same
|
|
// clock instead of actually waiting 320s+ of wall time. The judge stand-in records when each batch was
|
|
// judged (synchronous with RunOneBatch's own lastStart, since nothing sleeps between them) and cancels ctx
|
|
// once the 3rd has run, so RunBatches returns on its own next turn through the loop — this fails the moment
|
|
// RunBatches stops pacing, not only when nextBatchWait's own math would (TestNextBatchWait, above).
|
|
func TestThirtyWaitingMailsRunAsThreeBatchesAtLeast160sApart(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
store := newMemStore()
|
|
seedWaiting(t, store, 30)
|
|
seeded, err := store.Waiting(context.Background())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ids := make([]int64, len(seeded))
|
|
for i, w := range seeded {
|
|
ids[i] = w.Seq
|
|
}
|
|
|
|
var clockMu sync.Mutex
|
|
clock := time.Date(2026, 10, 20, 10, 0, 0, 0, time.UTC)
|
|
now := func() time.Time {
|
|
clockMu.Lock()
|
|
defer clockMu.Unlock()
|
|
return clock
|
|
}
|
|
sleep := func(d time.Duration) {
|
|
clockMu.Lock()
|
|
clock = clock.Add(d)
|
|
clockMu.Unlock()
|
|
}
|
|
|
|
var starts []time.Time
|
|
call := 0
|
|
judge := func(string, map[string]any) (json.RawMessage, HealthOutcome, string) {
|
|
starts = append(starts, now())
|
|
start := call * MaxMailsPerBatch
|
|
end := start + MaxMailsPerBatch
|
|
if end > len(ids) {
|
|
end = len(ids)
|
|
}
|
|
batch := ids[start:end]
|
|
call++
|
|
mails := make([]map[string]any, len(batch))
|
|
for i, id := range batch {
|
|
mails[i] = map[string]any{"id": id, "mark": "neither"}
|
|
}
|
|
raw, _ := json.Marshal(map[string]any{"mails": mails})
|
|
if len(starts) == 3 {
|
|
cancel()
|
|
}
|
|
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, noopLogf, MinBatchGap)
|
|
close(done)
|
|
}()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("RunBatches did not stop after its own 3rd batch and this test's cancel")
|
|
}
|
|
|
|
if len(starts) != 3 {
|
|
t.Fatalf("judge was called %d time(s), want 3 (30 mails / %d per batch)", len(starts), 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 previous, want at least %s", i, gap, MinBatchGap)
|
|
}
|
|
}
|
|
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)
|
|
}
|
|
}
|
|
|
|
// 1519/AC2, 1520/AC4 (review round 1 finding 4): the ask's own error is sorted into its real outcome by the
|
|
// exact words the real claude-code module answers with (modules/claude-code/cmd/claude-code/judge.go) —
|
|
// never OutcomeRefused for all three, which is what every ask error landed as before.
|
|
func TestNewJudgeSortsItsCauseByTheRealJudgesWords(t *testing.T) {
|
|
cases := []struct {
|
|
name string
|
|
err string
|
|
want HealthOutcome
|
|
}{
|
|
{"timed out", "the agent on novox did not answer within 40s; the judgement was ended", OutcomeTimedOut},
|
|
{"out of shape", `the agent on novox answered, and not in the shape asked: "mails" is not an array`, OutcomeOutOfShape},
|
|
{"refused", "the agent on novox could not judge: holds no licence", OutcomeRefused},
|
|
}
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
ask := func(string, map[string]any) (json.RawMessage, error) { return nil, fmt.Errorf("%s", c.err) }
|
|
judge := NewJudge(ask, "novox", time.Now)
|
|
_, outcome, why := judge("the question", AnswerSchema())
|
|
if outcome != c.want {
|
|
t.Fatalf("outcome = %q, want %q", outcome, c.want)
|
|
}
|
|
if why != c.err {
|
|
t.Fatalf("why = %q, want the ask's own error %q, unchanged", why, c.err)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// 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 }, noopLogf); 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 }, noopLogf); 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 }, noopLogf); 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, noopLogf); 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)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// 1520 (review round 1 finding 3): a mail-store.read that answers "holds no message" (the mail was moved
|
|
// or deleted since it was heard) takes that one mail out of the waiting queue and the batch goes on with
|
|
// the rest — never unhealthy (ADR 0459 §12: health is the judgement's).
|
|
func TestRunOneBatchTakesAGoneMailOutAndGoesOnWithTheRest(t *testing.T) {
|
|
ctx := context.Background()
|
|
store := newMemStore()
|
|
seedWaiting(t, store, 2)
|
|
waiting, _ := store.Waiting(ctx)
|
|
gone, stays := waiting[0], waiting[1]
|
|
|
|
readText := func(account, folder string, uid uint32) (string, error) {
|
|
if uid == gone.UID {
|
|
return "", fmt.Errorf("the account %s holds no message %d in %s", account, uid, folder)
|
|
}
|
|
return "the mail's text", nil
|
|
}
|
|
judge := func(string, map[string]any) (json.RawMessage, HealthOutcome, string) {
|
|
raw, _ := json.Marshal(map[string]any{"mails": []map[string]any{{"id": stays.Seq, "mark": "neither"}}})
|
|
return raw, OutcomeOK, ""
|
|
}
|
|
now := time.Date(2026, 10, 20, 11, 0, 0, 0, time.UTC)
|
|
if err := RunOneBatch(ctx, store, readText, judge, waiting, func() time.Time { return now }, noopLogf); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
entries, _ := store.Entries(ctx)
|
|
if len(entries) != 1 || entries[0].MessageID != stays.MessageID {
|
|
t.Fatalf("entries = %+v, want exactly one, for the mail that was still there", entries)
|
|
}
|
|
left, _ := store.Waiting(ctx)
|
|
if len(left) != 0 {
|
|
t.Fatalf("waiting = %v, want none (the gone mail taken out, the other judged)", left)
|
|
}
|
|
h, _ := store.ReaderHealth(ctx)
|
|
if !h.Healthy {
|
|
t.Fatalf("health = %+v, want healthy: a gone mail is never the judgement's problem", h)
|
|
}
|
|
}
|
|
|
|
// 1520 (review round 1 finding 3): any OTHER read failure leaves the whole batch's mails waiting, in
|
|
// order, for the next batch slot — the judge is never even asked — and never makes the reader unhealthy.
|
|
func TestRunOneBatchOtherReadFailureLeavesTheWholeBatchWaiting(t *testing.T) {
|
|
ctx := context.Background()
|
|
store := newMemStore()
|
|
seedWaiting(t, store, 2)
|
|
waiting, _ := store.Waiting(ctx)
|
|
|
|
readText := func(string, string, uint32) (string, error) {
|
|
return "", fmt.Errorf("the mailbox is unreachable")
|
|
}
|
|
judge := func(string, map[string]any) (json.RawMessage, HealthOutcome, string) {
|
|
t.Fatal("the judge must never be asked when a mail could not be read for a reason other than being gone")
|
|
return nil, OutcomeRefused, ""
|
|
}
|
|
now := time.Date(2026, 10, 20, 11, 0, 0, 0, time.UTC)
|
|
if err := RunOneBatch(ctx, store, readText, judge, waiting, func() time.Time { return now }, noopLogf); 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 {
|
|
t.Fatalf("health = %+v, want healthy: an ordinary read failure is never the judgement's problem", h)
|
|
}
|
|
}
|