novox/mesh-catalog
18
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

CAT-K6: a review waits for the agent module's judged event under its ticket; claude_code_judged is no longer invoked

Browse Source 
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
jschoubben
2026-11-04 12:41:50 +01:00
parent 4f433bdaf8
commit 8d4b2cb1c6
5 changed files with 164 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-knowledge/README.md
+2 -2
View File
@@ -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
modules/mesh-knowledge/cmd/mesh-knowledge/judge_bound_test.go
+32 -13
View File
@@ -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 {
modules/mesh-knowledge/cmd/mesh-knowledge/judged.go
+94
View File
@@ -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)
}
}
modules/mesh-knowledge/cmd/mesh-knowledge/main.go
+36 -10
View File
@@ -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
}
}
}
modules/mesh-knowledge/module.json
-1
View File
@@ -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",