diff --git a/modules/mesh-knowledge/README.md b/modules/mesh-knowledge/README.md index 55f45b49d..fc89568b3 100644 --- a/modules/mesh-knowledge/README.md +++ b/modules/mesh-knowledge/README.md @@ -109,8 +109,8 @@ as text. `get` serves each, and any other field a record of hq wrote under `fiel shows its state in the knowledge base apart from a research effort's own status (hq issue 492). The background work asks it at once, before the next step of any drafting, one judgement at a time. Every judgement asks a bound sized to its question: 25 seconds and 2 more per KiB, at most 120 (hq issue 508: an eleven-clause record was never - reviewed in 25). A judgement not done within one bus call is answered pending with a ticket, and collected from - the same node with `claude-code.claude_code_judged`. + reviewed in 25). A judgement not done within one bus call is answered pending with a ticket, and the review wakes + on the agent module's event `claude-code.judged` for that ticket (nothing asks again). - On `changes` the record stays proposed, the findings kept with the version they judged (`get` shows them). Its author revises it by proposing it again, and that version is reviewed again. - A judgement that cannot be had (the judge unreachable, or an answer that is not `approve`, or `changes` naming diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/judge_bound_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/judge_bound_test.go index e614a568c..b18da80a1 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/judge_bound_test.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/judge_bound_test.go @@ -68,7 +68,7 @@ func withinOf(t *testing.T, b map[string]any) time.Duration { func TestALongRecordsReviewGetsTheLargerBound(t *testing.T) { short, long := ReviewQuestion(aRecord(2, 200), nil), ReviewQuestion(aRecord(12, 2000), nil) a := &asked{} - j := judgementOver(a.ask, "novox") + j := judgementOver(a.ask, "novox", newJudgedHub()) for _, q := range []string{short, long} { if _, _, err := j(q, reviewSchema, "sonnet"); err != nil { t.Fatal(err) @@ -84,19 +84,38 @@ func TestALongRecordsReviewGetsTheLargerBound(t *testing.T) { } } -// A judgement answered pending is collected by its ticket from the node that holds it, and its answer is the -// judgement's. -func TestAPendingJudgementIsCollectedByItsTicket(t *testing.T) { - a := &asked{pending: 3} - answer, by, err := judgementOver(a.ask, "novox")(ReviewQuestion(aRecord(12, 2000), nil), reviewSchema, "sonnet") - if err != nil { - t.Fatal(err) +// A judgement answered pending waits for the agent module's word `judged` under its ticket, whether the word comes +// after the answer or before it, and asks nothing again; its error is the judgement's, and a word that never comes +// ends the wait in words. +func TestAPendingJudgementWaitsForTheWordOfItsTicket(t *testing.T) { + word := `{"ticket":"t1","node":"novox","model":"sonnet","answer":{"verdict":"approve","findings":[]}}` + for name, c := range map[string]struct{ before bool }{"after": {false}, "before": {true}} { + a := &asked{pending: 1} + hub := newJudgedHub() + if c.before { + hub.Heard(json.RawMessage(word)) + } else { + time.AfterFunc(50*time.Millisecond, func() { hub.Heard(json.RawMessage(word)) }) + } + answer, by, err := judgementOver(a.ask, "novox", hub)(ReviewQuestion(aRecord(12, 2000), nil), reviewSchema, "sonnet") + if err != nil { + t.Fatalf("%s: %v", name, err) + } + if string(answer) != `{"verdict":"approve","findings":[]}` || by != "the agent on novox (sonnet)" { + t.Fatalf("%s: %s by %s", name, answer, by) + } + if strings.Join(a.keys, " ") != "claude-code.claude_code_judge@novox" { + t.Fatalf("%s: asked %q", name, a.keys) + } } - if string(answer) != `{"verdict":"approve","findings":[]}` || by != "the agent on novox (sonnet)" { - t.Fatalf("%s by %s", answer, by) + hub := newJudgedHub() + hub.Heard(json.RawMessage(`{"ticket":"t1","node":"novox","error":"the agent on novox did not answer within 90s"}`)) + if _, _, err := judgementOver((&asked{pending: 1}).ask, "novox", hub)(ReviewQuestion(aRecord(12, 2000), nil), reviewSchema, "sonnet"); err == nil || + !strings.Contains(err.Error(), "did not answer within") { + t.Fatalf("the judgement's own failure: %v", err) } - if strings.Join(a.keys, " ") != "claude-code.claude_code_judge@novox claude-code.claude_code_judged@novox claude-code.claude_code_judged@novox claude-code.claude_code_judged@novox" { - t.Fatalf("asked %q", a.keys) + if _, err := newJudgedHub().Wait("novox", "t2", 20*time.Millisecond); err == nil || !strings.Contains(err.Error(), "t2") { + t.Fatalf("a word never said: %v", err) } } @@ -104,7 +123,7 @@ func TestAPendingJudgementIsCollectedByItsTicket(t *testing.T) { // it fits the old bound, and fails as before when it does not. func TestAnAgentModuleThatTakesNoWithinIsAskedWithoutIt(t *testing.T) { a := &asked{refuse: true} - if _, _, err := judgementOver(a.ask, "novox")(ReviewQuestion(aRecord(12, 2000), nil), reviewSchema, "sonnet"); err != nil { + if _, _, err := judgementOver(a.ask, "novox", newJudgedHub())(ReviewQuestion(aRecord(12, 2000), nil), reviewSchema, "sonnet"); err != nil { t.Fatal(err) } if len(a.bodies) != 2 || withinOf(t, a.bodies[1]) != 0 { diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/judged.go b/modules/mesh-knowledge/cmd/mesh-knowledge/judged.go new file mode 100644 index 000000000..891156a9d --- /dev/null +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/judged.go @@ -0,0 +1,94 @@ +package main + +// The agent module's word that a judgement ended (novox/hq ADR 0616 §5, issue 1956 note 8): a judgement asked with +// a bound past one bus call is answered pending with a ticket, and the module says `claude-code.judged` when the +// ticket's judgement ends. The knowledge base hears it here and the waiting review wakes; nothing asks the agent +// module again and again. + +import ( + "encoding/json" + "fmt" + "sync" + "time" +) + +// JudgedKept is how long a heard word is kept for a review that has not yet read its ticket (the word can arrive +// before the answer that names the ticket does). +const JudgedKept = 10 * time.Minute + +// judgedWord is the body of `claude-code.judged`. +type judgedWord struct { + Ticket string `json:"ticket"` + Node string `json:"node"` + Model string `json:"model"` + Answer json.RawMessage `json:"answer"` + Error string `json:"error"` +} + +type heardWord struct { + word judgedWord + at time.Time +} + +// judgedHub keeps the words heard, by node and ticket, and wakes whoever waits on one. +type judgedHub struct { + mu sync.Mutex + now func() time.Time + heard map[string]heardWord + waiting map[string][]chan struct{} +} + +func newJudgedHub() *judgedHub { + return &judgedHub{now: time.Now, heard: map[string]heardWord{}, waiting: map[string][]chan struct{}{}} +} + +func ticketKey(node, ticket string) string { return node + "/" + ticket } + +// Heard takes the body of one `judged` event. +func (h *judgedHub) Heard(body json.RawMessage) { + var w judgedWord + if json.Unmarshal(body, &w) != nil || w.Ticket == "" { + return + } + k := ticketKey(w.Node, w.Ticket) + h.mu.Lock() + now := h.now() + for key, o := range h.heard { + if now.Sub(o.at) > JudgedKept { + delete(h.heard, key) + } + } + h.heard[k] = heardWord{w, now} + waiters := h.waiting[k] + delete(h.waiting, k) + h.mu.Unlock() + for _, c := range waiters { + close(c) + } +} + +// Wait is the word for a ticket, waited for up to d: it is read once, and a ticket nobody said is an error that +// names the ticket. +func (h *judgedHub) Wait(node, ticket string, d time.Duration) (judgedWord, error) { + k := ticketKey(node, ticket) + h.mu.Lock() + if o, ok := h.heard[k]; ok { + delete(h.heard, k) + h.mu.Unlock() + return o.word, nil + } + c := make(chan struct{}) + h.waiting[k] = append(h.waiting[k], c) + h.mu.Unlock() + select { + case <-c: + h.mu.Lock() + o := h.heard[k] + delete(h.heard, k) + h.mu.Unlock() + return o.word, nil + case <-time.After(d): + return judgedWord{}, fmt.Errorf("the agent on %s did not say that ticket %s ended within %s; the review is asked again "+ + "(claude_code_judged on that node collects it by hand)", node, ticket, d) + } +} diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/main.go b/modules/mesh-knowledge/cmd/mesh-knowledge/main.go index e2c366d6c..0a5b09afd 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/main.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/main.go @@ -105,6 +105,7 @@ func newKB(store Store) *KB { func serve(k *KB, status string) { k.kick = make(chan struct{}, 1) go hear(k) + go hearJudged(heardJudged) go background(k) go ServePage(k) if err := stdio.Serve("", append(append(Tools(k, callerOf), healthTool(status, k.now, k), highestTool(k), marksTool(k)), moduleTools(k, callerOf)...)); err != nil { @@ -137,6 +138,22 @@ func hear(k *KB) { } } +// hearJudged subscribes to the agent modules' word that a judgement ended, asked again until the runtime takes it. +func hearJudged(h *judgedHub) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + err := stdio.Subscribe("claude-code.judged", func(e stdio.Envelope) error { + h.Heard(e.Body) + return nil + }) + if err == nil { + logf("[mesh-knowledge] hears when a judgement ends") + return + } + logf("[mesh-knowledge] does not hear when a judgement ends yet (%v); asking again in %s", err, wait) + time.Sleep(wait) + } +} + // background reads the router's record of every waiting ask once a minute, withdraws reservations never // proposed, reviews every proposal waiting for its review — at once when one is proposed — and drafts and // confirms clauses one step at a time, a review always before the next drafting step: the agent judges one @@ -219,7 +236,10 @@ func busReserver(kind, title string) (int, error) { } // busJudgement asks the agent module on this node for one judgement (ADR 0278). -func busJudgement(node string) Judgement { return judgementOver(stdio.Ask, node) } +func busJudgement(node string) Judgement { return judgementOver(stdio.Ask, node, heardJudged) } + +// heardJudged is what the agent modules have said ended (the event `claude-code.judged`, heard in hear). +var heardJudged = newJudgedHub() // The bound a judgement asks scales with its question (novox/hq issue 508): ADR 0297, eleven clauses and eleven // links, was never reviewed within the agent module's 25 seconds, while short records took 14 to 25. Each KiB of @@ -239,9 +259,9 @@ func JudgeWithin(question string) time.Duration { } // judgementOver asks the agent module through ask. A bound past one bus call's 30 seconds is answered pending with -// a ticket when the judgement is not done in the call, and collected from the node that holds it with -// claude_code_judged (novox/hq issue 508). -func judgementOver(ask func(key string, body any) (json.RawMessage, error), node string) Judgement { +// a ticket when the judgement is not done in the call, and its end is heard as the agent module's event `judged` +// (novox/hq issue 508, ADR 0616 §5); nothing asks for it again. +func judgementOver(ask func(key string, body any) (json.RawMessage, error), node string, hub *judgedHub) Judgement { return func(question string, schema map[string]any, model string) (json.RawMessage, string, error) { key := "claude-code.claude_code_judge" if node != "" { @@ -258,9 +278,6 @@ func judgementOver(ask func(key string, body any) (json.RawMessage, error), node delete(body, "within") raw, err = ask(key, body) } - // Past the judgement's bound the agent module ends it and says so; the deadline only guards against a - // module that answers pending for ever. - deadline := time.Now().Add(within + time.Minute) for { if err != nil { return nil, "", err @@ -284,10 +301,19 @@ func judgementOver(ask func(key string, body any) (json.RawMessage, error), node if out.Ticket == "" || out.Node == "" { return nil, "", fmt.Errorf("the agent module answered pending with no ticket or node: %.200s", raw) } - if time.Now().After(deadline) { - return nil, "", fmt.Errorf("the agent on %s still had not answered %s after its bound of %s", out.Node, out.Ticket, within) + // The agent module ends the judgement at its bound and says so in its word; this wait only guards + // against a word that never comes. + w, err := hub.Wait(out.Node, out.Ticket, within+time.Minute) + if err != nil { + return nil, "", err } - raw, err = ask("claude-code.claude_code_judged@"+out.Node, map[string]any{"ticket": out.Ticket}) + if w.Error != "" { + return nil, "", errors.New(w.Error) + } + if len(w.Answer) == 0 { + return nil, "", fmt.Errorf("the agent on %s said ticket %s ended with no answer", out.Node, out.Ticket) + } + return w.Answer, fmt.Sprintf("the agent on %s (%s)", out.Node, w.Model), nil } } } diff --git a/modules/mesh-knowledge/module.json b/modules/mesh-knowledge/module.json index 1da1278cb..f890fbf06 100644 --- a/modules/mesh-knowledge/module.json +++ b/modules/mesh-knowledge/module.json @@ -223,7 +223,6 @@ "invokes": [ "seat:issue-tracker.reserve", "claude-code.claude_code_judge", - "claude-code.claude_code_judged", "records.records_status", "records.records_list", "records.records_read",