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

K-A1: claude-code asks every question and proposal; the router holds the oldest back, the slot keys go

Browse Source 
Recovered from 90d70b83 (feat/47-G-asks).
Builds: design 47 G §40; design 46 §6 Limits, batching, history (no ask hidden or refused for its asker's count)

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-11 23:23:44 +01:00
parent 469763547c
commit 7df879b367
4 changed files with 17 additions and 277 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/claude-code/cmd/claude-code/proposals.go
-24
View File
@@ -906,10 +906,6 @@ func Propose(p Paths, in ProposeInput, store ProposalStore, bus AskBus, now time
if why := approverReady(bus); why != "" {
return refuse(why + "; " + AtTheTerminalInstead)
}
if waiting := waitingProposals(store, approved, now); waiting >= occ.MostOpen {
return refuse(fmt.Sprintf("%d proposals already wait for the operator's answer, and the operator is asked at most %d "+
"questions at once; claude_code_proposals lists them", waiting, occ.MostOpen))
}
if problem := ProposalProblem(prop); problem != "" {
return refuse(problem)
}
@@ -956,26 +952,6 @@ func Propose(p Paths, in ProposeInput, store ProposalStore, bus AskBus, now time
}, nil
}
// waitingProposals counts the proposals still open: asked, not decided here, not expired.
func waitingProposals(store ProposalStore, approved Approved, now time.Time) int {
keys, err := store.Keys()
if err != nil {
return 0
}
n := 0
for _, k := range keys {
prop, ok := readProposal(store, k)
if !ok || prop.Asked != "" {
continue
}
if _, decided := approved.Decided[prop.ID]; decided || now.After(prop.At.Add(ProposalLasts)) {
continue
}
n++
}
return n
}
func readProposal(store ProposalStore, key string) (Proposal, bool) {
raw, found, err := store.Get(key)
if err != nil || !found {
modules/claude-code/cmd/claude-code/questions.go
+3 -101
View File
@@ -21,7 +21,6 @@ import (
"errors"
"fmt"
"regexp"
"sort"
"strconv"
"strings"
"sync"
@@ -38,26 +37,8 @@ const QuestionsState = "questions"
// QuestionPrefix starts every question's key, so the router's word on one is never taken for a proposal's.
const QuestionPrefix = "q-"
// MostQuestions is how many questions may wait at once, mesh-wide: the router allows an asker three open asks
// (occ.MostOpen), and questions alone never fill them, so one is left for a proposed instruction section (ADR 0274)
// while two questions are open. Each open question holds one slot key in the `questions` state (slotKey), taken by
// create or by compare-and-set, so two nodes asking at once cannot both take the last one.
const MostQuestions = occ.MostOpen - 1
// A question the router holds no record of, and whose refusal was not heard here, keeps its slot until it expires:
// the router may still have it in its queue (down, or backlogged), and neither the queue's order nor a cancel can
// take it back — the queue offers a piece of work again after an error, so a cancel may be taken before the ask it
// cancels, and a cancel of an ask not yet recorded is dropped. Once the question has expired the router refuses it
// when it is taken (oc.Ask.Check), so the slot is free and the question has ended.
// slotKey is the key of one of the MostQuestions slots: not a question's key, so no question is ever read from it.
func slotKey(n int) string { return fmt.Sprintf("slot-%d", n) }
// slot is what a slot key holds: the question holding it.
type slot struct {
Question string `json:"question"`
At time.Time `json:"at"`
}
// A question is never refused for how many wait: the router offers an asker at most three asks at once and holds the
// oldest back behind newer ones, offered again when one ends (novox/hq ADR 0666 §12), so no question is hidden here.
// The bounds of a question's life (ADR 0332 §2): at least five minutes, at most a day less ten minutes (the
// router's bound for an approval, with the margin a proposal keeps), an hour when none is given.
@@ -372,19 +353,9 @@ func AskQuestion(p Paths, in QuestionInput, store QuestionStore, bus AskBus, now
if err := occ.Check(ask, now); err != nil {
return refuse(err.Error())
}
// The question is kept before its slot is taken, so a slot always names a question another node can judge.
if err := store.Create(q.ID, q); err != nil {
return nil, fmt.Errorf("the question could not be kept, so the operator was not asked: %w", err)
}
if open, err := takeSlot(store, bus, q, now); err != nil {
_ = store.Put(q.ID, tombstone(q, "not asked: "+err.Error()))
return refuse(err.Error())
} else if open != nil {
_ = store.Put(q.ID, tombstone(q, "not asked: two questions were open"))
return refuse(fmt.Sprintf("%d questions already wait for the operator's answer (%s), and sessions hold at most %d "+
"at once, so one of the router's three open asks is left for a proposed instruction section; read them with "+
"claude_code_answer, or wait for one to end", len(open), strings.Join(open, ", "), MostQuestions))
}
if err := bus.Publish(oc.AcceptSubjectAsk(Module), ask, ask.ID); err != nil {
_ = store.Put(q.ID, tombstone(q, "not asked: "+err.Error()))
return nil, fmt.Errorf("the operator could not be asked: %v", err)
@@ -420,75 +391,6 @@ func labelsOf(q Question) string {
return strings.Join(out, ", ")
}
// questionOpen says whether a question still waits: kept, asked, not ended, not expired, and open in the router's
// record or not recorded yet. An unreadable record is open: a slot is never freed on a guess.
func questionOpen(store QuestionStore, bus AskBus, id string, now time.Time) (bool, error) {
q, ok, err := readQuestionErr(store, id)
if err != nil {
return true, err
}
if !ok || q.Asked != "" || q.Ended != "" || !now.Before(q.Expires) {
return false, nil
}
entry, err := bus.Record(q.ID)
if err != nil {
return true, err
}
if entry == nil {
return true, nil
}
var rec routerRecord
if json.Unmarshal(entry.Value, &rec) != nil {
return true, nil
}
return rec.State == "" || rec.State == "open", nil
}
// takeSlot takes one of the MostQuestions slots for q: a slot no key holds, by create, or one whose question no longer
// waits, by compare-and-set at the revision read. Of two nodes taking the same slot only one wins; the other tries
// the next. It answers the open questions when every slot is held, and an error — refusing, never guessing — when a
// slot or its question cannot be read.
func takeSlot(store QuestionStore, bus AskBus, q Question, now time.Time) ([]string, error) {
mine := slot{Question: q.ID, At: q.At}
var open []string
for n := 1; n <= MostQuestions; n++ {
key := slotKey(n)
raw, rev, found, err := store.Get(key)
if err != nil {
return nil, fmt.Errorf("the open questions could not be counted (%v), so none more is asked", err)
}
if !found {
if err := store.Create(key, mine); err == nil {
return nil, nil
} else if !errors.Is(err, stdio.ErrChanged) {
return nil, fmt.Errorf("the open questions could not be counted (%v), so none more is asked", err)
}
if raw, rev, found, err = store.Get(key); err != nil || !found {
return nil, fmt.Errorf("the open questions could not be counted, so none more is asked")
}
}
var held slot
if json.Unmarshal(raw, &held) == nil && held.Question != "" {
waits, err := questionOpen(store, bus, held.Question, now)
if err != nil {
return nil, fmt.Errorf("the open questions could not be counted (%v), so none more is asked", err)
}
if waits {
open = append(open, held.Question)
continue
}
}
if err := store.Update(key, mine, rev); err == nil {
return nil, nil
} else if !errors.Is(err, stdio.ErrChanged) {
return nil, fmt.Errorf("the open questions could not be counted (%v), so none more is asked", err)
}
open = append(open, "one taken just now")
}
sort.Strings(open)
return open, nil
}
// tombstone is what is kept of a question that was not asked or that the router refused: its key, who asked, when, and
// why it ended — none of its words, which the router judged unfit to leave the mesh or never saw.
func tombstone(q Question, ended string) Question {
@@ -608,7 +510,7 @@ func ReadAnswer(store QuestionStore, bus AskBus, key string, now time.Time) (map
case entry == nil:
out["state"] = AnswerWaiting
out["words"] = "not yet recorded by the router: it may still be in the router's queue, or was refused without the " +
"refusal being heard here. It holds its slot, and the operator may still be asked, until it expires at " +
"refusal being heard here. The operator may still be asked until it expires at " +
q.Expires.Format(time.RFC3339)
case entry == nil, rec.State == "open", rec.Warrant == nil:
out["state"] = AnswerWaiting
modules/claude-code/cmd/claude-code/questions_test.go
+11 -146
View File
@@ -6,7 +6,6 @@ package main
import (
"encoding/json"
"errors"
"strings"
"sync"
"testing"
@@ -182,22 +181,21 @@ func TestAQuestionIsRefusedOutsideItsBounds(t *testing.T) {
}
}
func TestAtMostTwoQuestionsWaitSoAProposalFindsRoom(t *testing.T) {
// ADR 0666 §12: a question is never refused for how many wait; the router holds the oldest back, offered again when one ends.
func TestAThirdQuestionIsAskedNotRefusedAndAProposalIsAskedBesideThem(t *testing.T) {
p, _ := node(t, "g14")
store, r := newMemQuestions(), newRouter()
first := askMergeTonight(t, p, store, r, t0)
askMergeTonight(t, p, store, r, t0)
if _, err := AskQuestion(p, mergeTonight(), store, r, t0); err == nil || !strings.Contains(err.Error(), "2 questions already wait") {
t.Fatalf("a third question: %v", err)
for i := 0; i < 4; i++ {
if _, err := AskQuestion(p, mergeTonight(), store, r, t0); err != nil {
t.Fatalf("question %d: %v", i+1, err)
}
}
r.close(r.chose(first, "o1", t0.Add(time.Minute)))
if _, err := AskQuestion(p, mergeTonight(), store, r, t0.Add(2*time.Minute)); err != nil {
t.Fatalf("a question once one was answered: %v", err)
if len(r.subjects) != 4 {
t.Errorf("the router was asked %d times, want 4", len(r.subjects))
}
// A proposal is still asked beside two waiting questions.
pstore := &memProposals{m: map[string]json.RawMessage{}}
if _, err := Propose(p, ProposeInput{Scope: ScopeMesh, Name: "x", Text: "Text.\n"}, pstore, r, t0); err != nil {
t.Errorf("a proposal beside two questions: %v", err)
t.Errorf("a proposal beside the questions: %v", err)
}
}
@@ -234,9 +232,6 @@ func TestTheRoutersRefusalIsSaidAndKept(t *testing.T) {
if err != nil || out["state"] != AnswerEnded || !strings.Contains(out["words"].(string), "refused") {
t.Errorf("read after the refusal: %v %v", out, err)
}
if waits, err := questionOpen(store, refusing, refusing.id, t0.Add(time.Minute)); waits || err != nil {
t.Errorf("a refused question still counts as waiting: %v %v", waits, err)
}
var kept Question
_ = json.Unmarshal(store.m[refusing.id], &kept)
if kept.Headline != "" || kept.Context != "" || len(kept.Options) != 0 || kept.Why != "" || kept.OnExpiry != "" {
@@ -316,10 +311,6 @@ type memQuestions struct {
m map[string]json.RawMessage
rev map[string]uint64
n uint64
// failGet fails every read of a key with this prefix.
failGet string
// afterGet runs after a read, outside the lock: another node writing between a read and a write.
afterGet func(key string)
}
func newMemQuestions() *memQuestions {
@@ -328,17 +319,9 @@ func newMemQuestions() *memQuestions {
func (s *memQuestions) Get(key string) (json.RawMessage, uint64, bool, error) {
s.mu.Lock()
if s.failGet != "" && strings.HasPrefix(key, s.failGet) {
s.mu.Unlock()
return nil, 0, false, errors.New("the bus did not answer")
}
defer s.mu.Unlock()
raw, ok := s.m[key]
rev, after := s.rev[key], s.afterGet
s.mu.Unlock()
if after != nil {
after(key)
}
return raw, rev, ok, nil
return raw, s.rev[key], ok, nil
}
func (s *memQuestions) put(key string, value any) error {
raw, err := json.Marshal(value)
@@ -397,121 +380,3 @@ func TestAWarrantForAnotherAskOrOptionIsNotProven(t *testing.T) {
}
}
}
// Two nodes asking at the same moment take two slots at most: of three sessions racing, one is refused.
func TestTheCapHoldsAcrossNodesAskingAtOnce(t *testing.T) {
store, r := newMemQuestions(), &lockedRouter{router: newRouter()}
nodes := []string{"g14", "shanks", "ace"}
var wg sync.WaitGroup
errs := make([]error, len(nodes))
for i, name := range nodes {
p, _ := node(t, name)
wg.Add(1)
go func(i int, p Paths) {
defer wg.Done()
in := mergeTonight()
in.By = "fix-589"
_, errs[i] = AskQuestion(p, in, store, r, t0)
}(i, p)
}
wg.Wait()
refused := 0
for _, err := range errs {
if err != nil {
if !strings.Contains(err.Error(), "questions already wait") {
t.Errorf("refused for another reason: %v", err)
}
refused++
}
}
if refused != 1 {
t.Errorf("%d of three questions asked at once were refused; one is", refused)
}
}
// A slot that cannot be read refuses the question: the cap fails closed.
func TestTheCapFailsClosed(t *testing.T) {
p, _ := node(t, "g14")
store, r := newMemQuestions(), newRouter()
store.failGet = "slot-"
if _, err := AskQuestion(p, mergeTonight(), store, r, t0); err == nil || !strings.Contains(err.Error(), "could not be counted") {
t.Fatalf("asked with the slots unreadable: %v", err)
}
if len(r.subjects) != 0 {
t.Error("the operator was asked anyway")
}
}
// A question whose refusal was not heard here, and of which the router holds no record, keeps its slot and reads as
// waiting until it expires — the router may still take it from its queue — and only then frees its slot and ends.
func TestAQuestionTheRouterHasNotRecordedHoldsItsSlotUntilItExpires(t *testing.T) {
p, _ := node(t, "g14")
store, r := newMemQuestions(), newRouter()
first := askMergeTonight(t, p, store, r, t0)
askMergeTonight(t, p, store, r, t0)
delete(r.records, first) // the router down or backlogged, or its refusal never heard here
later := t0.Add(QuestionDefault - time.Minute)
if _, err := AskQuestion(p, mergeTonight(), store, r, later); err == nil {
t.Fatal("the slot of a question the router may still show was freed before it expired")
}
out, _ := ReadAnswer(store, r, first, later)
if out["state"] != AnswerWaiting || !strings.Contains(out["words"].(string), "not yet recorded") {
t.Errorf("before expiry read as %v", out)
}
expired := t0.Add(QuestionDefault)
if out, _ := ReadAnswer(store, r, first, expired); out["state"] != AnswerEnded ||
out["now"] != "do what you said: I leave the pull request open" {
t.Errorf("at expiry read as %v", out)
}
if _, err := AskQuestion(p, mergeTonight(), store, r, expired); err != nil {
t.Errorf("the slot of an expired question was not freed: %v", err)
}
}
// A slot is taken only at the revision read: when another node takes both slots between this node's read and its
// write, this node's writes are refused and the question is refused — a plain write would take a slot it lost.
func TestASlotIsTakenOnlyAtTheRevisionRead(t *testing.T) {
p, _ := node(t, "g14")
store, r := newMemQuestions(), newRouter()
// Two questions that ended: their slots are free to take again.
for _, id := range []string{askMergeTonight(t, p, store, r, t0), askMergeTonight(t, p, store, r, t0)} {
r.close(oc.Warrant{Ask: id, Outcome: occ.OutcomeExpired, At: occ.FormatTime(t0.Add(time.Minute))})
}
shanks, _ := node(t, "shanks")
other := newRouter()
store.afterGet = func(key string) {
if !strings.HasPrefix(key, "slot-") {
return
}
// Another node asks between this node's read of the slot and its write: its question waits.
q := Question{ID: newQuestionID(), Headline: "Another", Context: "c", Why: "w", OnExpiry: "e", By: "other",
From: shanks.Node, At: t0, Expires: t0.Add(time.Hour),
Options: []QuestionOption{{ID: "o1", Label: "A", Means: "a", Recommended: true}, {ID: "o2", Label: "B", Means: "b"}}}
_ = store.Put(q.ID, q)
_ = other.Publish(oc.AcceptSubjectAsk(Module), q.Ask(), q.ID)
r.records[q.ID] = other.records[q.ID]
_ = store.Put(key, slot{Question: q.ID, At: t0})
}
_, err := AskQuestion(p, mergeTonight(), store, r, t0.Add(2*time.Minute))
store.afterGet = nil
if err == nil || !strings.Contains(err.Error(), "questions already wait") {
t.Fatalf("a slot lost to another node was taken: %v", err)
}
}
// lockedRouter is the router fake, safe for sessions asking at once.
type lockedRouter struct {
mu sync.Mutex
*router
}
func (l *lockedRouter) Publish(subject string, body any, id string) error {
l.mu.Lock()
defer l.mu.Unlock()
return l.router.Publish(subject, body, id)
}
func (l *lockedRouter) Record(id string) (*stdio.StateEntry, error) {
l.mu.Lock()
defer l.mu.Unlock()
return l.router.Record(id)
}