95 lines
2.6 KiB
Go
95 lines
2.6 KiB
Go
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)
|
|
}
|
|
}
|