253 lines
10 KiB
Go
253 lines
10 KiB
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"
|
|
"time"
|
|
)
|
|
|
|
// BatchGap is how long must pass between the start of one batch and the next (ADR 0459 §15, 1520/AC1).
|
|
const BatchGap = 160 * time.Second
|
|
|
|
// JudgeWithin is the bound every judgement asks: never above 40 seconds (ADR 0459 §14, 1520/AC2).
|
|
const JudgeWithin = 40 * time.Second
|
|
|
|
// Ask is the one primitive every bus call in this file goes through — stdio.Ask in the running module
|
|
// (main.go), a recording stand-in in a test (1520/AC5).
|
|
type Ask func(key string, body map[string]any) (json.RawMessage, error)
|
|
|
|
// ReadTextFunc fetches one mail's text, never kept beyond the call that asked for it.
|
|
type ReadTextFunc func(account, folder string, uid uint32) (string, error)
|
|
|
|
// NewReadText is ReadTextFunc over seat:mail-store.read (ADR 0459 §3): the same seat mesh-mail-reader's
|
|
// manifest already invokes (1426/AC2).
|
|
func NewReadText(ask Ask) ReadTextFunc {
|
|
return func(account, folder string, uid uint32) (string, error) {
|
|
raw, err := ask("seat:mail-store.read", map[string]any{"account": account, "folder": folder, "uid": int(uid)})
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
var out struct {
|
|
Text string `json:"text"`
|
|
}
|
|
if err := json.Unmarshal(raw, &out); err != nil {
|
|
return "", fmt.Errorf("mail-store.read's answer is not a message: %.200s", raw)
|
|
}
|
|
return out.Text, nil
|
|
}
|
|
}
|
|
|
|
// JudgeFunc asks one batch's question and answers its outcome: the raw answer and OutcomeOK on success, or no
|
|
// answer and the outcome that says why (1520/AC4).
|
|
type JudgeFunc func(question string, schema map[string]any) (answer json.RawMessage, outcome HealthOutcome, why string)
|
|
|
|
// NewJudge is JudgeFunc over claude-code.claude_code_judge on this reader's own machine (the lead's decision
|
|
// at this milestone's prepare step: node is MESH_NODE), collecting a pending answer with claude_code_judged
|
|
// and its ticket until it answers or JudgeWithin plus a minute's grace has passed (1520/AC2) — the same shape
|
|
// mesh-knowledge's own judgementOver polls (cmd/mesh-knowledge/main.go). A bus call that itself errors (the
|
|
// judge module refused the ask outright, such as "holds no licence") is OutcomeRefused; an answer that keeps no
|
|
// shape ParseAnswer or this loop can read is OutcomeOutOfShape; running past the deadline is OutcomeTimedOut.
|
|
func NewJudge(ask Ask, node string, now func() time.Time) JudgeFunc {
|
|
key := "claude-code.claude_code_judge"
|
|
if node != "" {
|
|
key += "@" + node
|
|
}
|
|
return func(question string, schema map[string]any) (json.RawMessage, HealthOutcome, string) {
|
|
raw, err := ask(key, map[string]any{"question": question, "schema": schema, "within": int(JudgeWithin / time.Second)})
|
|
deadline := now().Add(JudgeWithin + time.Minute)
|
|
for {
|
|
if err != nil {
|
|
return nil, OutcomeRefused, err.Error()
|
|
}
|
|
var out struct {
|
|
Answer json.RawMessage `json:"answer"`
|
|
Node string `json:"node"`
|
|
Ticket string `json:"ticket"`
|
|
Pending bool `json:"pending"`
|
|
}
|
|
if jsonErr := json.Unmarshal(raw, &out); jsonErr != nil {
|
|
return nil, OutcomeOutOfShape, fmt.Sprintf("the judge's answer is not a judgement: %.200s", raw)
|
|
}
|
|
if !out.Pending {
|
|
if len(out.Answer) == 0 {
|
|
return nil, OutcomeOutOfShape, fmt.Sprintf("the judge's answer is not a judgement: %.200s", raw)
|
|
}
|
|
return out.Answer, OutcomeOK, ""
|
|
}
|
|
if out.Ticket == "" || out.Node == "" {
|
|
return nil, OutcomeOutOfShape, "the judge answered pending with no ticket or node"
|
|
}
|
|
if now().After(deadline) {
|
|
return nil, OutcomeTimedOut, fmt.Sprintf("no answer within %s", JudgeWithin)
|
|
}
|
|
raw, err = ask("claude-code.claude_code_judged@"+out.Node, map[string]any{"ticket": out.Ticket})
|
|
}
|
|
}
|
|
}
|
|
|
|
// nextBatchWait is how long to wait before the next batch may start (1520/AC1): zero once BatchGap has
|
|
// passed since the last start (or no batch has started yet), else the remainder.
|
|
func nextBatchWait(lastStart, now time.Time) time.Duration {
|
|
if lastStart.IsZero() {
|
|
return 0
|
|
}
|
|
elapsed := now.Sub(lastStart)
|
|
if elapsed >= BatchGap {
|
|
return 0
|
|
}
|
|
return BatchGap - elapsed
|
|
}
|
|
|
|
func senderOf(from []Person) (name, address string) {
|
|
if len(from) == 0 {
|
|
return "", ""
|
|
}
|
|
return from[0].Name, from[0].Address
|
|
}
|
|
|
|
func groupSummariesFrom(ctx context.Context, store Store) ([]GroupSummary, error) {
|
|
groups, err := store.Groups(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := make([]GroupSummary, len(groups))
|
|
for i, g := range groups {
|
|
out[i] = GroupSummary{Label: g.Label, Count: g.Count}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// recordOutcome folds one judgement's outcome into the reader's kept health (1519, 1520/AC4): writes no
|
|
// entry, and leaves whatever the caller already left alone.
|
|
func recordOutcome(ctx context.Context, store Store, outcome HealthOutcome, why string, now time.Time) error {
|
|
current, err := store.ReaderHealth(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return store.SetReaderHealth(ctx, RecordJudgement(current, outcome, why, now))
|
|
}
|
|
|
|
// RunOneBatch takes at most MaxMailsPerBatch of the waiting mails given (already in arrival order — the
|
|
// caller's to select, so a test can hand it an exact slice), builds the question (question.go) from them, the
|
|
// existing mail groups and the applicable reading lessons, judges it, and applies the answer:
|
|
//
|
|
// - on success (1520/AC3): one mail entry per mail, any new mail group found-or-created by its label, and
|
|
// each mail leaves the waiting queue;
|
|
// - on a refused, timed-out or out-of-shape judgement (1520/AC4): no entry is written, every given mail
|
|
// stays waiting, in the same order, and the reader's health (1519) is set unhealthy with the outcome and
|
|
// why; the next successful batch sets it healthy again.
|
|
//
|
|
// mesh-mail-reader never asks to send, move, flag or delete (ADR 0459 §11): this function asks only readText
|
|
// (seat:mail-store.read) and judge (claude-code.claude_code_judge, claude_code_judged) — 1520/AC5 is proven by
|
|
// building both from a recording Ask and checking what it recorded, not by anything this function does itself.
|
|
func RunOneBatch(ctx context.Context, store Store, readText ReadTextFunc, judge JudgeFunc, waiting []WaitingMail, now func() time.Time) error {
|
|
if len(waiting) > MaxMailsPerBatch {
|
|
waiting = waiting[:MaxMailsPerBatch]
|
|
}
|
|
if len(waiting) == 0 {
|
|
return nil
|
|
}
|
|
lessons, err := store.Lessons(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
groups, err := groupSummariesFrom(ctx, store)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
mails := make([]QuestionMail, len(waiting))
|
|
ids := make([]int64, len(waiting))
|
|
for i, w := range waiting {
|
|
text, err := readText(w.Account, w.Folder, w.UID)
|
|
if err != nil {
|
|
// mail-store.read itself failing stops the batch exactly as a refused judgement would: the mails
|
|
// stay waiting, and the reader is made unhealthy with the cause.
|
|
return recordOutcome(ctx, store, OutcomeRefused, "reading the mail's text: "+err.Error(), now())
|
|
}
|
|
mails[i] = QuestionMail{ID: w.Seq, Account: w.Account, Folder: w.Folder, Date: w.Date, From: w.From, Subject: w.Subject, Flags: w.Flags, Text: text}
|
|
ids[i] = w.Seq
|
|
}
|
|
question := BuildQuestion(mails, groups, lessons)
|
|
raw, outcome, why := judge(question.Text, AnswerSchema())
|
|
if outcome != OutcomeOK {
|
|
return recordOutcome(ctx, store, outcome, why, now())
|
|
}
|
|
answer, err := ParseAnswer(raw, ids)
|
|
if err != nil {
|
|
return recordOutcome(ctx, store, OutcomeOutOfShape, err.Error(), now())
|
|
}
|
|
judgedAt := now()
|
|
for _, w := range waiting {
|
|
j := answer.Mails[w.Seq]
|
|
var groupID int64
|
|
if j.Group != "" {
|
|
g, err := store.FindOrCreateGroup(ctx, j.Group)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
groupID = g.ID
|
|
}
|
|
var due time.Time
|
|
if j.DueDate != "" {
|
|
due, _ = time.Parse("2006-01-02", j.DueDate) // ParseAnswer already proved this parses
|
|
}
|
|
name, address := senderOf(w.From)
|
|
e := Entry{
|
|
Account: w.Account, MessageID: w.MessageID, Folder: w.Folder, Date: w.Date,
|
|
SenderName: name, SenderAddress: address, Subject: w.Subject,
|
|
Group: groupID, Summary: j.Summary, Mark: j.Mark, Why: j.Why, DueDate: due, Judged: judgedAt,
|
|
}
|
|
if _, err := store.AddEntry(ctx, e); err != nil {
|
|
return err
|
|
}
|
|
if err := store.TakeWaiting(ctx, w.Seq); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return recordOutcome(ctx, store, OutcomeOK, "", judgedAt)
|
|
}
|
|
|
|
// RunBatches is the loop the running module drives (main.go): pace batch starts at least BatchGap apart
|
|
// (nextBatchWait), take at most MaxMailsPerBatch of the waiting mails in arrival order, and run one batch at a
|
|
// time, forever, until ctx is done. Not unit-tested itself beyond compiling (go vet): 1520/AC1's pacing and
|
|
// AC2-AC5's one-batch behaviour are each proven directly, on nextBatchWait and RunOneBatch, in batch_test.go.
|
|
func RunBatches(ctx context.Context, store Store, readText ReadTextFunc, judge JudgeFunc, now func() time.Time, sleep func(time.Duration), logf func(string, ...any)) {
|
|
var lastStart time.Time
|
|
for {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
if wait := nextBatchWait(lastStart, now()); wait > 0 {
|
|
sleep(wait)
|
|
continue
|
|
}
|
|
waiting, err := store.Waiting(ctx)
|
|
if err != nil {
|
|
logf("[mesh-mail-reader] the waiting mails could not be read: %v", err)
|
|
sleep(time.Second)
|
|
continue
|
|
}
|
|
if len(waiting) == 0 {
|
|
sleep(time.Second)
|
|
continue
|
|
}
|
|
if len(waiting) > MaxMailsPerBatch {
|
|
waiting = waiting[:MaxMailsPerBatch]
|
|
}
|
|
lastStart = now()
|
|
if err := RunOneBatch(ctx, store, readText, judge, waiting, now); err != nil {
|
|
logf("[mesh-mail-reader] a batch failed: %v", err)
|
|
}
|
|
}
|
|
}
|