diff --git a/modules/claude-code/README.md b/modules/claude-code/README.md index 9ac99a29..44973ad8 100644 --- a/modules/claude-code/README.md +++ b/modules/claude-code/README.md @@ -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 diff --git a/modules/claude-code/cmd/claude-code/judge.go b/modules/claude-code/cmd/claude-code/judge.go index 435be4c9..724cfbd0 100644 --- a/modules/claude-code/cmd/claude-code/judge.go +++ b/modules/claude-code/cmd/claude-code/judge.go @@ -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) + }} +} diff --git a/modules/claude-code/cmd/claude-code/judge_test.go b/modules/claude-code/cmd/claude-code/judge_test.go index b5bce4e7..0ad535d0 100644 --- a/modules/claude-code/cmd/claude-code/judge_test.go +++ b/modules/claude-code/cmd/claude-code/judge_test.go @@ -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") + } +} diff --git a/modules/claude-code/cmd/claude-code/main.go b/modules/claude-code/cmd/claude-code/main.go index b7c59927..95d5188d 100644 --- a/modules/claude-code/cmd/claude-code/main.go +++ b/modules/claude-code/cmd/claude-code/main.go @@ -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)...) } diff --git a/modules/claude-code/module.json b/modules/claude-code/module.json index 5a4c1890..8be60f2e 100644 --- a/modules/claude-code/module.json +++ b/modules/claude-code/module.json @@ -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": [ diff --git a/modules/mesh-knowledge/README.md b/modules/mesh-knowledge/README.md index 4657b39c..020e1ffb 100644 --- a/modules/mesh-knowledge/README.md +++ b/modules/mesh-knowledge/README.md @@ -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. diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/judge_bound_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/judge_bound_test.go new file mode 100644 index 00000000..e614a568 --- /dev/null +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/judge_bound_test.go @@ -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) + } +} diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/main.go b/modules/mesh-knowledge/cmd/mesh-knowledge/main.go index b2836798..02a7a349 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/main.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/main.go @@ -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 }} } diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/manifest_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/manifest_test.go index de2550f4..d9d5b069 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/manifest_test.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/manifest_test.go @@ -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) } diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/review.go b/modules/mesh-knowledge/cmd/mesh-knowledge/review.go index 4a58c4ce..917810f4 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/review.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/review.go @@ -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) + } } } diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/review_bounds_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/review_bounds_test.go index 974d5635..8f30a968 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/review_bounds_test.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/review_bounds_test.go @@ -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) } } diff --git a/modules/mesh-knowledge/module.json b/modules/mesh-knowledge/module.json index 47ceed56..dbb08b00 100644 --- a/modules/mesh-knowledge/module.json +++ b/modules/mesh-knowledge/module.json @@ -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",