Review a long record within a bound sized to it, and let a proposal waiting on its author leave the knowledge base healthy (hq issue 508)
mesh/merge-gate pass: builds claude-code, mesh-knowledge → ace, g14, novox, shanks; no bus step; every machine composes with the change as it did without (…
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-check-typescript.sh (typescript)
mesh/delivery delivered
mesh/merge-gate pass: builds claude-code, mesh-knowledge → ace, g14, novox, shanks; no bus step; every machine composes with the change as it did without (…
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-check-typescript.sh (typescript)
mesh/delivery delivered
A judgement over one bus call's 30 seconds is collected by its ticket from claude-code, so an eleven-clause record can be reviewed; a proposal out of review tries is told to the operator, not raised as the module's fault, so it no longer holds the module's own rollouts.
This commit is contained in:
@@ -199,7 +199,10 @@ only a language model can, for another module — the issue tracker's duplicate
|
||||
its own), in an empty directory, the question on its input: no built-in tools, no tool servers, plugins, hooks or
|
||||
instruction files, no session kept, so it can only answer. It answers the structured answer, the model, the node
|
||||
and the seconds, within 25 seconds — the process is ended past that, under the bus's 30 — and one at a time on a
|
||||
node. Refused: an empty question or one over 64 KiB, a schema that is not an object, a model not `sonnet`, `haiku`
|
||||
node. A caller may ask a longer bound with `within` (seconds, at most 120), for a long question (hq issue 508): a
|
||||
judgement not done in 20 seconds is then answered `pending` with a `ticket`, runs on to its bound, and
|
||||
`claude_code_judged` (`ticket`) collects it, waiting up to 20 seconds a call. A ticket is answered once; an answer
|
||||
nobody collects is dropped ten minutes after its bound, and none survives a restart of the module. Refused: an empty question or one over 64 KiB, a schema that is not an object, a model not `sonnet`, `haiku`
|
||||
or `opus`, and a node holding no licence. The question is never logged.
|
||||
|
||||
## Settings
|
||||
|
||||
@@ -12,12 +12,15 @@ package main
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
@@ -27,6 +30,14 @@ const (
|
||||
// JudgeBound is how long one judgement may take, waiting for another included: under the bus's 30 seconds
|
||||
// a tool call is answered in, so the caller hears this module's refusal rather than a timeout.
|
||||
JudgeBound = 25 * time.Second
|
||||
// MostJudgeBound is the longest bound a caller may ask with within: a long record's review takes longer than
|
||||
// one bus call allows (novox/hq issue 508), and is then collected by its ticket.
|
||||
MostJudgeBound = 120 * time.Second
|
||||
// JudgeWait is how long one call waits for a judgement asked with a bound past JudgeBound before it answers
|
||||
// the ticket: well under the bus's 30 seconds, so the call is always answered.
|
||||
JudgeWait = 20 * time.Second
|
||||
// JudgedKept is how long an answer nobody collected is kept past its judgement's bound.
|
||||
JudgedKept = 10 * time.Minute
|
||||
// MostQuestion is the largest question taken.
|
||||
MostQuestion = 64 << 10
|
||||
// MostSchema is the largest answer schema taken.
|
||||
@@ -43,14 +54,29 @@ type JudgeAsk struct {
|
||||
Question string
|
||||
Schema map[string]any
|
||||
Model string
|
||||
// Within is the bound the caller asks; JudgeBound when zero. Past JudgeBound the call answers a ticket if the
|
||||
// judgement is not done within JudgeWait, and claude_code_judged collects it (novox/hq issue 508).
|
||||
Within time.Duration
|
||||
}
|
||||
|
||||
// Judged is one answer.
|
||||
type Judged struct {
|
||||
Answer json.RawMessage `json:"answer"`
|
||||
Answer json.RawMessage `json:"answer,omitempty"`
|
||||
Model string `json:"model"`
|
||||
Node string `json:"node"`
|
||||
Seconds float64 `json:"seconds"`
|
||||
// Ticket names a judgement asked with a bound past JudgeBound; Pending says it is not done yet, and
|
||||
// claude_code_judged with the ticket collects it.
|
||||
Ticket string `json:"ticket,omitempty"`
|
||||
Pending bool `json:"pending,omitempty"`
|
||||
}
|
||||
|
||||
// running is one judgement asked with a bound past JudgeBound, until it is collected.
|
||||
type running struct {
|
||||
done chan struct{}
|
||||
out *Judged
|
||||
err error
|
||||
until time.Time
|
||||
}
|
||||
|
||||
// Judge runs judgements on this node, one at a time.
|
||||
@@ -63,7 +89,13 @@ type Judge struct {
|
||||
Now func() time.Time
|
||||
// Bound is how long one judgement may take; JudgeBound unless a test says less.
|
||||
Bound time.Duration
|
||||
turn chan struct{}
|
||||
// Wait is how long a call waits for a long judgement before it answers its ticket; JudgeWait unless a test
|
||||
// says less.
|
||||
Wait time.Duration
|
||||
turn chan struct{}
|
||||
|
||||
mu sync.Mutex
|
||||
tickets map[string]*running
|
||||
}
|
||||
|
||||
// NewJudge is the judge of this node.
|
||||
@@ -118,11 +150,25 @@ func (j *Judge) Judge(a JudgeAsk) (*Judged, error) {
|
||||
"bindings say which node holds which", j.P.Node)
|
||||
}
|
||||
|
||||
start := j.Now()
|
||||
bound := j.Bound
|
||||
if bound <= 0 {
|
||||
bound = JudgeBound
|
||||
}
|
||||
if a.Within > MostJudgeBound {
|
||||
return nil, fmt.Errorf("within is %s; a judgement takes at most %s", a.Within, MostJudgeBound)
|
||||
}
|
||||
if a.Within > bound {
|
||||
return j.start(string(schema), model, question, a.Within)
|
||||
}
|
||||
if a.Within > 0 {
|
||||
bound = a.Within
|
||||
}
|
||||
return j.run(string(schema), model, question, bound)
|
||||
}
|
||||
|
||||
// run is one judgement within its bound, waiting for its turn included.
|
||||
func (j *Judge) run(schema, model, question string, bound time.Duration) (*Judged, error) {
|
||||
start := j.Now()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), bound)
|
||||
defer cancel()
|
||||
select {
|
||||
@@ -131,7 +177,7 @@ func (j *Judge) Judge(a JudgeAsk) (*Judged, error) {
|
||||
case <-ctx.Done():
|
||||
return nil, fmt.Errorf("another judgement held %s's agent for all of %s; ask again", j.P.Node, bound)
|
||||
}
|
||||
out, err := j.Exec(ctx, j.Argv(string(schema), model), []byte(question))
|
||||
out, err := j.Exec(ctx, j.Argv(schema, model), []byte(question))
|
||||
if ctx.Err() != nil {
|
||||
return nil, fmt.Errorf("the agent on %s did not answer within %s; the judgement was ended", j.P.Node, bound)
|
||||
}
|
||||
@@ -145,6 +191,68 @@ func (j *Judge) Judge(a JudgeAsk) (*Judged, error) {
|
||||
return &Judged{Answer: answer, Model: model, Node: j.P.Node, Seconds: j.Now().Sub(start).Round(100 * time.Millisecond).Seconds()}, nil
|
||||
}
|
||||
|
||||
// start runs a judgement whose bound is past what one bus call allows, and waits for it as long as a call may: its
|
||||
// answer when it is done by then, its ticket otherwise (novox/hq issue 508).
|
||||
func (j *Judge) start(schema, model, question string, bound time.Duration) (*Judged, error) {
|
||||
var b [12]byte
|
||||
_, _ = rand.Read(b[:])
|
||||
ticket := hex.EncodeToString(b[:])
|
||||
r := &running{done: make(chan struct{}), until: j.Now().Add(bound + JudgedKept)}
|
||||
j.mu.Lock()
|
||||
j.forget()
|
||||
if j.tickets == nil {
|
||||
j.tickets = map[string]*running{}
|
||||
}
|
||||
j.tickets[ticket] = r
|
||||
j.mu.Unlock()
|
||||
go func() {
|
||||
r.out, r.err = j.run(schema, model, question, bound)
|
||||
close(r.done)
|
||||
}()
|
||||
return j.Collect(ticket)
|
||||
}
|
||||
|
||||
// Collect waits as long as a call may for the judgement under a ticket: its answer, why it failed, or that it is
|
||||
// still pending. A ticket is answered once; an answer nobody collects is kept JudgedKept past its bound.
|
||||
func (j *Judge) Collect(ticket string) (*Judged, error) {
|
||||
j.mu.Lock()
|
||||
j.forget()
|
||||
r := j.tickets[ticket]
|
||||
j.mu.Unlock()
|
||||
if r == nil {
|
||||
return nil, fmt.Errorf("no judgement on %s under ticket %q: it was collected, kept past its time, or this module "+
|
||||
"restarted since; ask the judgement again", j.P.Node, ticket)
|
||||
}
|
||||
wait := j.Wait
|
||||
if wait <= 0 {
|
||||
wait = JudgeWait
|
||||
}
|
||||
select {
|
||||
case <-r.done:
|
||||
case <-time.After(wait):
|
||||
return &Judged{Node: j.P.Node, Ticket: ticket, Pending: true}, nil
|
||||
}
|
||||
j.mu.Lock()
|
||||
delete(j.tickets, ticket)
|
||||
j.mu.Unlock()
|
||||
if r.err != nil {
|
||||
return nil, r.err
|
||||
}
|
||||
out := *r.out
|
||||
out.Ticket = ticket
|
||||
return &out, nil
|
||||
}
|
||||
|
||||
// forget drops the answers nobody collected in time. Called with mu held.
|
||||
func (j *Judge) forget() {
|
||||
now := j.Now()
|
||||
for t, r := range j.tickets {
|
||||
if now.After(r.until) {
|
||||
delete(j.tickets, t)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// answerOf reads print mode's result: the structured answer, or why there is none.
|
||||
func answerOf(out []byte) (json.RawMessage, error) {
|
||||
var r struct {
|
||||
@@ -213,17 +321,21 @@ func judgeTool(j *Judge) stdio.Tool {
|
||||
"object); optionally the model (" + strings.Join(JudgeModels, ", ") + "; default " + JudgeModels[0] + "). The agent " +
|
||||
"answers as the account agents run as here, under the licence this node is bound to, with no tools, " +
|
||||
"no tool servers, no instructions and no session kept — it can only answer — within 25 seconds or " +
|
||||
"not at all. One at a time on a node. Refused: an empty or larger than 64 KiB question, a schema that " +
|
||||
"not at all, unless within asks longer (at most 120 seconds, for a long question): then, when it is not done in " +
|
||||
"20 seconds, it answers pending with a ticket, and claude_code_judged collects the answer (novox/hq issue 508). " +
|
||||
"One at a time on a node. Refused: an empty or larger than 64 KiB question, a schema that " +
|
||||
"is not an object, an unknown model, a node holding no licence. The question is never logged.",
|
||||
Input: map[string]any{
|
||||
"question": str("the question, with everything the answer needs"),
|
||||
"schema": map[string]any{"type": "object", "description": "the JSON schema the answer follows"},
|
||||
"model": map[string]any{"type": "string", "enum": JudgeModels, "description": "the model (default " + JudgeModels[0] + ")"},
|
||||
"within": map[string]any{"type": "number", "description": "the bound in seconds, at most 120 (default 25); " +
|
||||
"past 25 a judgement not done in 20 seconds answers pending with a ticket for claude_code_judged"},
|
||||
},
|
||||
Run: func(a map[string]any) (any, error) {
|
||||
for k := range a {
|
||||
if k != "question" && k != "schema" && k != "model" {
|
||||
return nil, fmt.Errorf("claude_code_judge takes question, schema and model; not %s", k)
|
||||
if k != "question" && k != "schema" && k != "model" && k != "within" {
|
||||
return nil, fmt.Errorf("claude_code_judge takes question, schema, model and within; not %s", k)
|
||||
}
|
||||
}
|
||||
var ask JudgeAsk
|
||||
@@ -241,6 +353,35 @@ func judgeTool(j *Judge) stdio.Tool {
|
||||
return nil, errors.New("model is text")
|
||||
}
|
||||
}
|
||||
if v, given := a["within"]; given && v != nil {
|
||||
secs, ok := v.(float64)
|
||||
if !ok || secs <= 0 {
|
||||
return nil, errors.New("within is a number of seconds")
|
||||
}
|
||||
ask.Within = time.Duration(secs * float64(time.Second))
|
||||
}
|
||||
return j.Judge(ask)
|
||||
}}
|
||||
}
|
||||
|
||||
// judgedTool collects a judgement asked with a bound past one bus call's (novox/hq issue 508).
|
||||
func judgedTool(j *Judge) stdio.Tool {
|
||||
return stdio.Tool{Name: "claude_code_judged",
|
||||
Description: "The answer of a judgement claude_code_judge answered pending with a ticket (novox/hq issue 508): " +
|
||||
"waits up to 20 seconds, then answers the judgement's answer, why it failed, or pending again. A ticket is " +
|
||||
"answered once; one nobody collects is dropped ten minutes after its bound, and none survives a restart of " +
|
||||
"this module.",
|
||||
Input: map[string]any{"ticket": str("the ticket claude_code_judge answered")},
|
||||
Run: func(a map[string]any) (any, error) {
|
||||
for k := range a {
|
||||
if k != "ticket" {
|
||||
return nil, fmt.Errorf("claude_code_judged takes ticket; not %s", k)
|
||||
}
|
||||
}
|
||||
ticket, ok := a["ticket"].(string)
|
||||
if !ok || ticket == "" {
|
||||
return nil, errors.New("claude_code_judged needs ticket, as claude_code_judge answered it")
|
||||
}
|
||||
return j.Collect(ticket)
|
||||
}}
|
||||
}
|
||||
|
||||
@@ -166,3 +166,102 @@ func TestTheToolTakesOnlyItsArguments(t *testing.T) {
|
||||
t.Fatalf("%s", raw)
|
||||
}
|
||||
}
|
||||
|
||||
// A long question may ask a longer bound than one bus call allows (novox/hq issue 508): the call answers a ticket
|
||||
// when the judgement is not done within the call's wait, the judgement runs on to its own bound, and
|
||||
// claude_code_judged collects its answer.
|
||||
func TestALongJudgementIsCollectedByItsTicket(t *testing.T) {
|
||||
var deadline time.Time
|
||||
j := testJudge(true, true, func(ctx context.Context, _ []string, _ []byte) ([]byte, error) {
|
||||
deadline, _ = ctx.Deadline()
|
||||
select {
|
||||
case <-time.After(400 * time.Millisecond):
|
||||
return []byte(`{"structured_output":{"verdict":"none"}}`), nil
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
})
|
||||
j.Wait = 50 * time.Millisecond
|
||||
start := time.Now()
|
||||
out, err := j.Judge(JudgeAsk{Question: "a long record", Schema: verdictSchema, Within: 90 * time.Second})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !out.Pending || out.Ticket == "" || len(out.Answer) != 0 || out.Node != "control" {
|
||||
t.Fatalf("not a ticket: %+v", out)
|
||||
}
|
||||
var got *Judged
|
||||
for i := 0; i < 100; i++ {
|
||||
got, err = j.Collect(out.Ticket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !got.Pending {
|
||||
break
|
||||
}
|
||||
}
|
||||
if got.Pending || string(got.Answer) != `{"verdict":"none"}` || got.Ticket != out.Ticket {
|
||||
t.Fatalf("%+v", got)
|
||||
}
|
||||
// Read once the judgement is collected: its end is what orders the agent's write before this read.
|
||||
if left := deadline.Sub(start); left < 80*time.Second || left > 91*time.Second {
|
||||
t.Fatalf("the judgement's bound is %s, not the 90s asked", left)
|
||||
}
|
||||
if _, err := j.Collect(out.Ticket); err == nil || !strings.Contains(err.Error(), "no judgement") {
|
||||
t.Fatalf("a ticket collected twice: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// A long judgement that fails says why when it is collected, and its bound is never past MostJudgeBound.
|
||||
func TestALongJudgementIsBoundedAndSaysWhyItFailed(t *testing.T) {
|
||||
j := testJudge(true, true, func(ctx context.Context, _ []string, _ []byte) ([]byte, error) {
|
||||
<-ctx.Done()
|
||||
return nil, ctx.Err()
|
||||
})
|
||||
if _, err := j.Judge(JudgeAsk{Question: "q", Schema: verdictSchema, Within: MostJudgeBound + time.Second}); err == nil ||
|
||||
!strings.Contains(err.Error(), "at most") {
|
||||
t.Fatalf("a bound past the most taken: %v", err)
|
||||
}
|
||||
j.Wait = 20 * time.Millisecond
|
||||
j.Bound = 30 * time.Millisecond // the test's stand-in for JudgeBound: within is past it, so a ticket
|
||||
out, err := j.Judge(JudgeAsk{Question: "q", Schema: verdictSchema, Within: 200 * time.Millisecond})
|
||||
if err != nil || !out.Pending {
|
||||
t.Fatalf("%+v %v", out, err)
|
||||
}
|
||||
for i := 0; i < 100; i++ {
|
||||
got, err := j.Collect(out.Ticket)
|
||||
if err != nil {
|
||||
if !strings.Contains(err.Error(), "did not answer within") {
|
||||
t.Fatalf("%v", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if !got.Pending {
|
||||
t.Fatalf("answered: %+v", got)
|
||||
}
|
||||
}
|
||||
t.Fatal("never ended")
|
||||
}
|
||||
|
||||
// The tools take within and ticket, and nothing else.
|
||||
func TestTheToolsTakeWithinAndTicket(t *testing.T) {
|
||||
j := testJudge(true, true, func(context.Context, []string, []byte) ([]byte, error) {
|
||||
return []byte(`{"structured_output":{"verdict":"none"}}`), nil
|
||||
})
|
||||
out, err := judgeTool(j).Run(map[string]any{"question": "q", "schema": verdictSchema, "within": float64(60)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := out.(*Judged); got.Pending || string(got.Answer) != `{"verdict":"none"}` {
|
||||
t.Fatalf("a quick long judgement is answered in its call: %+v", got)
|
||||
}
|
||||
if _, err := judgeTool(j).Run(map[string]any{"question": "q", "schema": verdictSchema, "within": "long"}); err == nil {
|
||||
t.Fatal("within as text was taken")
|
||||
}
|
||||
if _, err := judgedTool(j).Run(map[string]any{"ticket": "nope"}); err == nil || !strings.Contains(err.Error(), "no judgement") {
|
||||
t.Fatalf("%v", err)
|
||||
}
|
||||
if _, err := judgedTool(j).Run(map[string]any{"ticket": "x", "by": "y"}); err == nil {
|
||||
t.Fatal("an unknown argument was taken")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -355,7 +355,8 @@ func tools(p Paths, servers ServerState, view *ServerView, config ConfigState, c
|
||||
Description: "Read-only: everything registered through this module that a Claude Code session runs or obeys — the tool servers, the hooks, the settings and the skills, subagents, commands, output styles and instructions — by key, with who registered it, when, and whether it is still rendered (novox/hq issue 340), beside what the operator set for this node at the controller's terminal (managed_settings, mcp_servers). `by` is the node whose tool runner served the call, never the caller.",
|
||||
Run: func(map[string]any) (any, error) { return Registrations(p, servers, config) }})
|
||||
out = append(out, proposalTools(p)...)
|
||||
out = append(out, judgeTool(NewJudge(p)))
|
||||
judge := NewJudge(p)
|
||||
out = append(out, judgeTool(judge), judgedTool(judge))
|
||||
return append(out, configTools(p, config, configView)...)
|
||||
}
|
||||
|
||||
|
||||
@@ -67,7 +67,8 @@
|
||||
"claude_code_home_remove",
|
||||
"claude_code_home_removed",
|
||||
"claude_code_home_restore",
|
||||
"claude_code_judge"
|
||||
"claude_code_judge",
|
||||
"claude_code_judged"
|
||||
],
|
||||
"data": {
|
||||
"own": [
|
||||
|
||||
@@ -99,7 +99,10 @@ as text. `get` serves each, and any other field a record of hq wrote under `fiel
|
||||
the record, the records it links and the rules, and answers `approve` or `changes` with findings. The rules it sees
|
||||
name every kind of link `propose` accepts, from the one list `propose` holds a link to, and the record it sees
|
||||
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.
|
||||
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`.
|
||||
- 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
|
||||
@@ -108,8 +111,9 @@ as text. `get` serves each, and any other field a record of hq wrote under `fiel
|
||||
version in the store and said in its history, so the background work takes it up at that time, also after a
|
||||
restart of the tool runner (hq issue 497). After 3 tries the record says it was the last and its author
|
||||
proposes it again to have it reviewed again; the operator is told once on the operator's channel (keyed by
|
||||
record and version, and cleared once the record is proposed again, accepted, withdrawn or superseded), and
|
||||
the module is unhealthy, naming it, so the mesh raises a condition.
|
||||
record and version, cleared once the record is proposed again, accepted, withdrawn or superseded, and sent
|
||||
again at the next step when it could not be sent). The module's health names the record and stays healthy:
|
||||
the record waits on its author, not on the module, so it never holds the module's own rollouts (hq issue 508).
|
||||
An ask of the operator that cannot be made counts as a try the same way.
|
||||
- A review whose outcome the database refused to keep is held back for that record alone (10, then 20
|
||||
minutes, 3 tries, logged as given up), so it never blocks the records behind it.
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// novox/hq issue 508: ADR 0297, eleven clauses and eleven links, could not be reviewed within the judgement's 25
|
||||
// seconds five times running, while short records were reviewed in 14 to 25. A review's bound scales with what is
|
||||
// judged, up to a cap, and a judgement past one bus call's time is collected by its ticket.
|
||||
|
||||
// aRecord is a decision of n clauses, each about size bytes.
|
||||
func aRecord(n, size int) *Record {
|
||||
r := &Record{Kind: KindDecision, Number: 297, Title: "Issues are kept as a log of operations", Status: StatusProposed}
|
||||
for i := 1; i <= n; i++ {
|
||||
r.Clauses = append(r.Clauses, Clause{N: i, Text: fmt.Sprintf("**Clause %d.** %s", i, strings.Repeat("A word of reasoning. ", size/20))})
|
||||
}
|
||||
r.Sections = []Section{{Name: "Context", Text: strings.Repeat("Why it was decided. ", n*size/40)}}
|
||||
return r
|
||||
}
|
||||
|
||||
// asked is the agent module, played over the calls the knowledge base makes to it.
|
||||
type asked struct {
|
||||
keys []string
|
||||
bodies []map[string]any
|
||||
pending int // how many times a ticketed judgement answers pending before its answer
|
||||
refuse bool
|
||||
}
|
||||
|
||||
func (a *asked) ask(key string, body any) (json.RawMessage, error) {
|
||||
b, _ := body.(map[string]any)
|
||||
a.keys, a.bodies = append(a.keys, key), append(a.bodies, b)
|
||||
if strings.HasPrefix(key, "claude-code.claude_code_judge@") {
|
||||
if _, long := b["within"]; long && a.refuse {
|
||||
return nil, errors.New("claude_code_judge takes question, schema and model; not within")
|
||||
}
|
||||
if _, long := b["within"]; long && a.pending > 0 {
|
||||
return json.RawMessage(`{"node":"novox","ticket":"t1","pending":true}`), nil
|
||||
}
|
||||
return json.RawMessage(`{"answer":{"verdict":"approve","findings":[]},"model":"sonnet","node":"novox"}`), nil
|
||||
}
|
||||
if key == "claude-code.claude_code_judged@novox" && b["ticket"] == "t1" {
|
||||
if a.pending--; a.pending > 0 {
|
||||
return json.RawMessage(`{"node":"novox","ticket":"t1","pending":true}`), nil
|
||||
}
|
||||
return json.RawMessage(`{"answer":{"verdict":"approve","findings":[]},"model":"sonnet","node":"novox","ticket":"t1"}`), nil
|
||||
}
|
||||
return nil, fmt.Errorf("nothing answers %s", key)
|
||||
}
|
||||
|
||||
func withinOf(t *testing.T, b map[string]any) time.Duration {
|
||||
t.Helper()
|
||||
v, ok := b["within"]
|
||||
if !ok {
|
||||
return 0
|
||||
}
|
||||
s, ok := v.(int)
|
||||
if !ok {
|
||||
t.Fatalf("within is not whole seconds: %#v", v)
|
||||
}
|
||||
return time.Duration(s) * time.Second
|
||||
}
|
||||
|
||||
func TestALongRecordsReviewGetsTheLargerBound(t *testing.T) {
|
||||
short, long := ReviewQuestion(aRecord(2, 200), nil), ReviewQuestion(aRecord(12, 2000), nil)
|
||||
a := &asked{}
|
||||
j := judgementOver(a.ask, "novox")
|
||||
for _, q := range []string{short, long} {
|
||||
if _, _, err := j(q, reviewSchema, "sonnet"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
s, l := withinOf(t, a.bodies[0]), withinOf(t, a.bodies[1])
|
||||
if l <= s || l <= 25*time.Second || l > MostJudgeWithin {
|
||||
t.Fatalf("a %d-byte review is bounded %s and a %d-byte one %s: the longer must get the larger bound, past 25s and at most %s",
|
||||
len(short), s, len(long), l, MostJudgeWithin)
|
||||
}
|
||||
if got := JudgeWithin(strings.Repeat("x", MostReviewQuestion)); got != MostJudgeWithin {
|
||||
t.Fatalf("the largest question is bounded %s, not the cap %s", got, MostJudgeWithin)
|
||||
}
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
if string(answer) != `{"verdict":"approve","findings":[]}` || by != "the agent on novox (sonnet)" {
|
||||
t.Fatalf("%s by %s", answer, by)
|
||||
}
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// An agent module from before issue 508, which takes no within, is asked again without it: the review is had when
|
||||
// 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 {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(a.bodies) != 2 || withinOf(t, a.bodies[1]) != 0 {
|
||||
t.Fatalf("asked %v", a.bodies)
|
||||
}
|
||||
}
|
||||
@@ -202,25 +202,76 @@ 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 {
|
||||
func busJudgement(node string) Judgement { return judgementOver(stdio.Ask, node) }
|
||||
|
||||
// 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
|
||||
// question adds JudgePerKiB to JudgeFloor, up to MostJudgeWithin, which the agent module takes at most. A sized
|
||||
// bound was chosen over asking clause by clause: every judgement pays the agent's start, which alone is most of
|
||||
// those 14 to 25 seconds, so a dozen clause-sized judgements would each risk the same 25 seconds, take minutes
|
||||
// together, and still need a judgement of the whole record for what binds its clauses together.
|
||||
const (
|
||||
JudgeFloor = 25 * time.Second
|
||||
JudgePerKiB = 2 * time.Second
|
||||
MostJudgeWithin = 120 * time.Second
|
||||
)
|
||||
|
||||
// JudgeWithin is the bound a judgement of this question asks.
|
||||
func JudgeWithin(question string) time.Duration {
|
||||
return min(JudgeFloor+time.Duration(len(question)>>10)*JudgePerKiB, MostJudgeWithin)
|
||||
}
|
||||
|
||||
// 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 {
|
||||
return func(question string, schema map[string]any, model string) (json.RawMessage, string, error) {
|
||||
key := "claude-code.claude_code_judge"
|
||||
if node != "" {
|
||||
key += "@" + node
|
||||
}
|
||||
raw, err := stdio.Ask(key, map[string]any{"question": question, "schema": schema, "model": model})
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
within := JudgeWithin(question)
|
||||
body := map[string]any{"question": question, "schema": schema, "model": model}
|
||||
if within > JudgeFloor {
|
||||
body["within"] = int(within / time.Second)
|
||||
}
|
||||
var out struct {
|
||||
Answer json.RawMessage `json:"answer"`
|
||||
Model string `json:"model"`
|
||||
Node string `json:"node"`
|
||||
raw, err := ask(key, body)
|
||||
if err != nil && body["within"] != nil && strings.Contains(err.Error(), "not within") {
|
||||
// An agent module from before issue 508 takes no within: asked again without it, under its own bound.
|
||||
delete(body, "within")
|
||||
raw, err = ask(key, body)
|
||||
}
|
||||
if err := json.Unmarshal(raw, &out); err != nil || len(out.Answer) == 0 {
|
||||
return nil, "", fmt.Errorf("the agent module's answer is not a judgement: %.200s", raw)
|
||||
// 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
|
||||
}
|
||||
var out struct {
|
||||
Answer json.RawMessage `json:"answer"`
|
||||
Model string `json:"model"`
|
||||
Node string `json:"node"`
|
||||
Ticket string `json:"ticket"`
|
||||
Pending bool `json:"pending"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &out); err != nil {
|
||||
return nil, "", fmt.Errorf("the agent module's answer is not a judgement: %.200s", raw)
|
||||
}
|
||||
if !out.Pending {
|
||||
if len(out.Answer) == 0 {
|
||||
return nil, "", fmt.Errorf("the agent module's answer is not a judgement: %.200s", raw)
|
||||
}
|
||||
return out.Answer, fmt.Sprintf("the agent on %s (%s)", out.Node, out.Model), nil
|
||||
}
|
||||
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)
|
||||
}
|
||||
raw, err = ask("claude-code.claude_code_judged@"+out.Node, map[string]any{"ticket": out.Ticket})
|
||||
}
|
||||
return out.Answer, fmt.Sprintf("the agent on %s (%s)", out.Node, out.Model), nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -501,9 +552,10 @@ type HealthAnswer struct {
|
||||
At time.Time `json:"at,omitempty"`
|
||||
}
|
||||
|
||||
// Health is how the knowledge base is: unhealthy when its database cannot serve, when a rule that judges the
|
||||
// whole record set fails, or when a proposed record ran out of review tries — the mesh raises a condition naming
|
||||
// the rule, the records, or the records not reviewed.
|
||||
// Health is whether the knowledge base works: unhealthy when its database cannot serve, or when a rule that judges
|
||||
// the whole record set fails — the mesh raises a condition naming the rule and the records. A proposal that ran out
|
||||
// of review tries waits on its author, not on the module: it is told to its author and the operator, and named
|
||||
// here, but leaves the module healthy, so it never holds the module's own rollouts (novox/hq issue 508).
|
||||
func (k *KB) Health(ready Healther) HealthAnswer {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
|
||||
defer cancel()
|
||||
@@ -525,33 +577,38 @@ func (k *KB) Health(ready Healther) HealthAnswer {
|
||||
if err != nil {
|
||||
return HealthAnswer{State: "unhealthy", Why: "the records cannot be read: " + err.Error()}
|
||||
}
|
||||
unreviewed := Unreviewed(all)
|
||||
if len(findings)+len(unreviewed) > 0 {
|
||||
var parts []string
|
||||
if len(unreviewed) > 0 {
|
||||
if len(unreviewed) > 8 {
|
||||
unreviewed = append(unreviewed[:8], fmt.Sprintf("and %d more", len(unreviewed)-8))
|
||||
}
|
||||
parts = append(parts, fmt.Sprintf("%d proposed record(s) could not be reviewed in %d tries and wait for their authors to propose them again: %s",
|
||||
len(Unreviewed(all)), MostReviewTries, strings.Join(unreviewed, "; ")))
|
||||
waiting := ""
|
||||
if unreviewed := Unreviewed(all); len(unreviewed) > 0 {
|
||||
n := len(unreviewed)
|
||||
if n > 8 {
|
||||
unreviewed = append(unreviewed[:8], fmt.Sprintf("and %d more", n-8))
|
||||
}
|
||||
if len(says) > 0 {
|
||||
if len(says) > 8 {
|
||||
says = append(says[:8], fmt.Sprintf("and %d more", len(says)-8))
|
||||
}
|
||||
parts = append(parts, fmt.Sprintf("%d rule(s) that judge the whole record set fail: %s", len(findings), strings.Join(says, "; ")))
|
||||
}
|
||||
return HealthAnswer{State: "unhealthy", Why: strings.Join(parts, ". ")}
|
||||
waiting = fmt.Sprintf("%d proposed record(s) could not be reviewed in %d tries and wait for their authors to propose them again: %s",
|
||||
n, MostReviewTries, strings.Join(unreviewed, "; "))
|
||||
}
|
||||
return HealthAnswer{State: "healthy", Healthy: true, Why: "the database serves, every rule that judges the whole record set holds, and no proposal ran out of review tries"}
|
||||
if len(says) > 0 {
|
||||
if len(says) > 8 {
|
||||
says = append(says[:8], fmt.Sprintf("and %d more", len(says)-8))
|
||||
}
|
||||
why := fmt.Sprintf("%d rule(s) that judge the whole record set fail: %s", len(findings), strings.Join(says, "; "))
|
||||
if waiting != "" {
|
||||
why += ". " + waiting
|
||||
}
|
||||
return HealthAnswer{State: "unhealthy", Why: why}
|
||||
}
|
||||
why := "the database serves and every rule that judges the whole record set holds"
|
||||
if waiting != "" {
|
||||
why += "; " + waiting
|
||||
}
|
||||
return HealthAnswer{State: "healthy", Healthy: true, Why: why}
|
||||
}
|
||||
|
||||
func healthTool(status string, now func() time.Time) stdio.Tool {
|
||||
return stdio.Tool{Name: HealthTool,
|
||||
Description: "How the knowledge base is, as its watch last saw it: healthy when its database serves this version and every " +
|
||||
"rule that judges the whole record set holds; unhealthy otherwise, naming the rule and the records — a decision accepted " +
|
||||
"that nothing cites, a supersession or a change said on one end only, a record extending a superseded one — or when a " +
|
||||
"proposed record ran out of review tries, naming it.",
|
||||
"that nothing cites, a supersession or a change said on one end only, a record extending a superseded one. A " +
|
||||
"proposed record that ran out of review tries is named but leaves it healthy: it waits on its author (novox/hq issue 508).",
|
||||
Run: func(map[string]any) (any, error) { return ReadStatus(status, now()), nil }}
|
||||
}
|
||||
|
||||
|
||||
@@ -76,7 +76,7 @@ func TestItDeclaresAndHoldsTheKnowledgeBaseSeatOnceForTheMesh(t *testing.T) {
|
||||
if !reflect.DeepEqual(m.Uses, []string{"operator-channel"}) || len(m.Consumes) != 0 {
|
||||
t.Fatalf("uses %v, consumes %v", m.Uses, m.Consumes)
|
||||
}
|
||||
if !reflect.DeepEqual(m.Invokes, []string{"seat:issue-tracker.reserve", "claude-code.claude_code_judge", "records.records_status",
|
||||
if !reflect.DeepEqual(m.Invokes, []string{"seat:issue-tracker.reserve", "claude-code.claude_code_judge", "claude-code.claude_code_judged", "records.records_status",
|
||||
"records.records_list", "records.records_read", "seat:operator-channel.notify"}) {
|
||||
t.Fatalf("invokes %v", m.Invokes)
|
||||
}
|
||||
|
||||
@@ -12,8 +12,9 @@ package main
|
||||
// and says so on it. It is never taken as an approval. It is owed: the try count and when the next try is due
|
||||
// are kept with the version in the store, so the background work takes it up at that time, also after a restart
|
||||
// of the tool runner (novox/hq issue 497), up to MostReviewTries. After the last try the record says so, its
|
||||
// author proposes it again to have it reviewed again, the module's health names it, and the operator is told
|
||||
// once on the operator's channel.
|
||||
// author proposes it again to have it reviewed again, and the operator is told once on the operator's channel.
|
||||
// The module's health names it and stays healthy: the record waits on its author, and the module works
|
||||
// (novox/hq issue 508).
|
||||
// - A record with a version whose writer the runtime did not prove is not reviewed: who wrote it cannot be told,
|
||||
// and no one but its proven author revises it, so its author proposes it again under a new number.
|
||||
// - On approve, a record that does not need the operator is accepted at once; one its author marked
|
||||
@@ -278,7 +279,7 @@ func (k *KB) ReviewStep() (string, error) {
|
||||
}
|
||||
|
||||
// Unreviewed are the proposed records that ran out of review tries: no judgement could be had, or the operator
|
||||
// could not be asked. They make the module unhealthy, so the mesh raises a condition naming them.
|
||||
// could not be asked. The module's health names them and stays healthy (novox/hq issue 508); the operator is told.
|
||||
func Unreviewed(all []*Entry) []string {
|
||||
var out []string
|
||||
for _, e := range all {
|
||||
@@ -326,19 +327,23 @@ func (k *KB) tellUnreviewed(e *Entry) {
|
||||
version = e.PendingVersion
|
||||
}
|
||||
key := fmt.Sprintf("%sv%d", unreviewedOf(w.Key()), version)
|
||||
// Kept as told before the notice is sent, as for drafted clauses: a notice that fails is logged and not sent
|
||||
// again, and the module's health still names the record, so the mesh's condition stands for it.
|
||||
// Kept as told before the notice is sent, so two steps never send it twice; a notice that fails is no longer
|
||||
// kept as told, and is sent again at the next step: the module's health stays healthy for it (novox/hq issue
|
||||
// 508), so this notice is the only loud word of it.
|
||||
if first, err := k.Store.Told(ctx, key+" "+at); err != nil || !first {
|
||||
return
|
||||
}
|
||||
summary := fmt.Sprintf("%s was never reviewed: its review could not be had %d times (last: %s). It stays proposed, and nothing of it is in force.",
|
||||
w.Key(), w.Judging.Tries, w.Judging.Reason)
|
||||
explanation := "No review that could not be had counts as an approval. Its author proposes it again to have it reviewed again; " +
|
||||
"until then the knowledge base's health names it. The knowledge base's get and history show it."
|
||||
"the knowledge base's get and history show it."
|
||||
// The channel edits a cleared message with the resolved line given here, not the one sent with the clear.
|
||||
resolved := w.Key().String() + " no longer waits for its review"
|
||||
if err := k.Notify(key, summary, "A proposal was never reviewed: "+w.Key().String(), explanation, resolved); err != nil {
|
||||
k.logf("[mesh-knowledge] the operator could not be told that %s was never reviewed: %v; the module's health still names it", w.Key(), err)
|
||||
k.logf("[mesh-knowledge] the operator could not be told that %s was never reviewed: %v; told again at the next step", w.Key(), err)
|
||||
if uerr := k.Store.Untold(ctx, key+" "+at); uerr != nil {
|
||||
k.logf("[mesh-knowledge] and that it was not told could not be kept: %v", uerr)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -73,21 +73,36 @@ func TestARecordWhoseReviewCannotBeKeptBlocksNoOtherAndIsTriedBoundedly(t *testi
|
||||
}
|
||||
}
|
||||
|
||||
// A record that ran out of review tries makes the module unhealthy, naming it: the mesh raises a condition.
|
||||
func TestARecordThatCouldNotBeReviewedMakesTheModuleUnhealthy(t *testing.T) {
|
||||
// A record that ran out of review tries waits on its author, not on the module: it is told to the operator and leaves
|
||||
// the module healthy, so it never holds the module's own rollouts (novox/hq issue 508). A notice that could not be
|
||||
// sent is sent again at the next step, since health no longer stands for it.
|
||||
func TestARecordThatCouldNotBeReviewedLeavesTheModuleHealthyAndIsTold(t *testing.T) {
|
||||
w := imported(t, newMemStore())
|
||||
n := proposed(t, w, "author@g14", false)
|
||||
w.judge.reviewFail = true
|
||||
for i := 0; i < MostReviewTries; i++ {
|
||||
notify := w.k.Notify
|
||||
failing := true
|
||||
w.k.Notify = func(key, summary, headline, explanation, resolved string) error {
|
||||
if failing {
|
||||
failing = false
|
||||
return errors.New("the operator's channel did not answer")
|
||||
}
|
||||
return notify(key, summary, headline, explanation, resolved)
|
||||
}
|
||||
for i := 0; i < MostReviewTries+1; i++ {
|
||||
w.reviewAll(t)
|
||||
if h := w.k.Health(nil); i < MostReviewTries-1 && !h.Healthy {
|
||||
t.Fatalf("unhealthy while tries remain: %s", h.Why)
|
||||
if h := w.k.Health(nil); !h.Healthy {
|
||||
t.Fatalf("unhealthy after try %d: %s", i+1, h.Why)
|
||||
}
|
||||
w.clock = w.clock.Add(time.Hour)
|
||||
}
|
||||
w.reviewAll(t)
|
||||
h := w.k.Health(nil)
|
||||
if h.Healthy || !strings.Contains(h.Why, itoaKey(n)) || !strings.Contains(h.Why, "could not be reviewed") {
|
||||
t.Fatalf("%+v", h)
|
||||
if !h.Healthy || h.State != "healthy" || !strings.Contains(h.Why, itoaKey(n)) || !strings.Contains(h.Why, "author") {
|
||||
t.Fatalf("not healthy, or the record waiting on its author not named: %+v", h)
|
||||
}
|
||||
if len(w.told) != 1 || !strings.Contains(w.told[0], itoaKey(n)) || !strings.Contains(w.told[0], "never reviewed") {
|
||||
t.Fatalf("told %q", w.told)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -71,6 +71,7 @@
|
||||
"invokes": [
|
||||
"seat:issue-tracker.reserve",
|
||||
"claude-code.claude_code_judge",
|
||||
"claude-code.claude_code_judged",
|
||||
"records.records_status",
|
||||
"records.records_list",
|
||||
"records.records_read",
|
||||
|
||||
Reference in New Issue
Block a user