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

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"
	"strings"
	"time"
)

// MinBatchGap is the pace no production path may ever lower (ADR 0459 §15, 1520/AC1): at least 160s
// between the start of one batch and the next. It is a const, never read by anything an integration
// build could shorten; the unit tests in batch_test.go assert against this literal, not against the var
// below, so they keep proving 160s even when compiled into a binary whose BatchGap has been shortened
// (issue 1520, fixed per the lead's try feedback on bb594652).
const MinBatchGap = 160 * time.Second

// BatchGap is the gap the running binary actually paces its batches by: a var, never a setting, so an
// integration test can shorten it for its own build alone (issue 1431's own fix: batch_gap_integration.go,
// `//go:build integration`, never part of a production build) without production pacing ever becoming
// configurable. Every normal (non-integration) build starts it at MinBatchGap and never changes it;
// TestRunningBatchGapIsMinBatchGap (batch_gap_test.go, `//go:build !integration`) proves that for the
// binary actually shipped, not just for a literal.
var BatchGap = MinBatchGap

// 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 {
				outcome, why := classifyJudgeError(err)
				return nil, outcome, why
			}
			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})
		}
	}
}

// classifyJudgeError sorts an error the judge's own ask (claude_code_judge, claude_code_judged) returned
// into its real outcome (review round 1 finding 4, issue 1519/AC2, 1520/AC4): the real claude-code module
// answers "did not answer within <bound>; the judgement was ended" for a timeout and "answered, and not in
// the shape asked: <why>" for a misshapen answer (modules/claude-code/cmd/claude-code/judge.go), both as
// this ask's own error, not as a JSON body NewJudge's loop above could read a Pending/Answer shape from —
// before this, every one of those landed as OutcomeRefused, indistinguishable from the agent module
// actually refusing the ask ("holds no licence"). Anything else, including that real refusal, is
// OutcomeRefused, as before.
func classifyJudgeError(err error) (HealthOutcome, string) {
	msg := err.Error()
	switch {
	case strings.Contains(msg, "did not answer within"):
		return OutcomeTimedOut, msg
	case strings.Contains(msg, "not in the shape asked"):
		return OutcomeOutOfShape, msg
	default:
		return OutcomeRefused, msg
	}
}

// isNoMessageError says whether err is mail-store.read's own "the account <a> holds no message <uid> in
// <folder>" (mesh-mail's cmd/mesh-mail/tools.go readTool): the mail named by account/folder/uid is no
// longer there — moved or deleted since it was heard — the one read failure RunOneBatch treats as the mail
// being simply gone rather than a reading problem (review round 1 finding 3, issue 1520).
func isNoMessageError(err error) bool {
	return err != nil && strings.Contains(err.Error(), "holds no message")
}

// nextBatchWait is how long to wait before the next batch may start (1520/AC1): zero once gap has
// passed since the last start (or no batch has started yet), else the remainder. gap is given explicitly
// by the caller (RunBatches passes BatchGap; the unit tests in batch_test.go pass MinBatchGap directly)
// so this pure function never itself reaches into a var an integration build might have shortened.
func nextBatchWait(lastStart, now time.Time, gap time.Duration) time.Duration {
	if lastStart.IsZero() {
		return 0
	}
	elapsed := now.Sub(lastStart)
	if elapsed >= gap {
		return 0
	}
	return gap - 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:
//
//   - a mail whose read answers "holds no message" (moved or deleted since it was heard, review round 1
//     finding 3, issue 1520) is taken out of the waiting queue, with a log line, and never reaches the
//     question — the batch goes on with the rest;
//   - any other read failure leaves every mail of this batch waiting, in the same order, for the next
//     batch slot — the whole batch, not only the one that failed to read;
//   - 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.
//
// Neither a mail that is simply gone nor an ordinary read failure ever makes the reader unhealthy (ADR 0459
// §12: health is the judgement's, review round 1 finding 3) — only a judgement's own outcome does.
//
// 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, logf func(string, ...any)) 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
	}
	live := make([]WaitingMail, 0, len(waiting))
	mails := make([]QuestionMail, 0, len(waiting))
	ids := make([]int64, 0, len(waiting))
	for _, w := range waiting {
		text, err := readText(w.Account, w.Folder, w.UID)
		if err != nil {
			if isNoMessageError(err) {
				if takeErr := store.TakeWaiting(ctx, w.Seq); 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)
				continue
			}
			// Any other read failure: the whole batch — this mail and whatever is after it — waits for
			// the next batch slot, and the reader stays whatever health it already was.
			logf("[mesh-mail-reader] reading %s %s uid %d: %v; this batch waits for the next slot", w.Account, w.Folder, w.UID, err)
			return nil
		}
		live = append(live, w)
		mails = append(mails, 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 = append(ids, w.Seq)
	}
	if len(live) == 0 {
		// Every mail given was gone: the queue already reflects it, and there is nothing left to judge.
		return nil
	}
	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 live {
		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 gap apart
// (nextBatchWait), take at most MaxMailsPerBatch of the waiting mails in arrival order, 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
// integration` build ever shortens; TestThirtyWaitingMailsRunAsThreeBatchesAtLeast160sApart passes
// MinBatchGap) so this function itself never reads that var — the try on 8000b637 found this test still
// reading it (and so, compiled into the integration binary, pacing at the shortened 3s instead of 160s)
// 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) {
	var lastStart time.Time
	for {
		if ctx.Err() != nil {
			return
		}
		if wait := nextBatchWait(lastStart, now(), gap); 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, 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)
		}
	}
}