diff --git a/modules/claude-code/README.md b/modules/claude-code/README.md index 16dbccac0..965787d04 100644 --- a/modules/claude-code/README.md +++ b/modules/claude-code/README.md @@ -83,7 +83,7 @@ must see, a node that joins later included — kept, so it carries no secret eit | what this node should hold | the licence manager's `bindings` state, this node's key; on a newer generation this module asks `anthropic-licence-manager.current` for its token, sealed to the key it sends, and writes it access-token-only — so the agent here never refreshes. A node that was off reads its key when it is back | | an MCP server registered through this module before registering was refused (hq issue 340) | a key in the module's `servers` state — `all.` for every node, `.` for one; every node watches it and renders what applies to it, a node's own entry over the one for every node. A node that joins later, or was off, reads the whole current set at start; unregistering is a delete. Nothing adds a key any more: a new tool server is the operator's, in the `mcp_servers` setting at the controller's terminal | | a proposed instruction section (hq ADR 0274) | a key in the module's `proposals` state per proposal, the ask published on `operator-channel` under this module's name, and the router's record of it read under that name alone; what a node approved is its own file, never the bus | -| a session's question to the operator (hq ADR 0332) | a key `q-…` in the module's `questions` state per question, the ask published on `operator-channel` under this module's name, and the router's record of it read under that name alone; the router's word that it ended without a choice is kept on the question; two slot keys `slot-1`, `slot-2` name the questions open | +| a session's question to the operator (hq ADR 0332) | a key `q-…` in the module's `questions` state per question, the ask published on `operator-channel` under this module's name, and the router's record of it read under that name alone; the router's word that it ended without a choice is kept on the question | | a project's memory (hq ADR 0343) | a key in the module's `memory` state per file — `.`, `operator` for the operator's own homes and `agent` for an agent account's — holding the text, its digest, the digest of the version it replaced, the node and when it changed; a removal is a value marked removed, never a deleted key. Written at the revision read, so of two nodes writing one file only one wins. Every node of the same kind watches it. No event and no tool answer carries memory text | | the agent's configuration (hq ADR 0216) | a key in the module's `config` state per registration — `mesh..` for every node, `node...` for one, `home...` for one account's own directory — the item and its files in one value, at most 256 KiB. Every node watches it and renders what applies to it, a node item over a mesh item of the same kind and name. A hook, settings, or an item that would run something by itself is never taken, whenever it was put there (hq issue 340) | @@ -233,11 +233,8 @@ key says waiting, chosen (the option, who answered, on which channel, when), end refused, cancelled, replaced), or not proven: the answer is read from the router's own record and held to the question as `questions` keeps it, so a question rewritten after it was asked, a warrant for another ask or a choice after expiry is never read as a choice. One call waits at most 60 s; the tool asks a session to read a question at -most once a minute, and nothing enforces it. At most two questions wait at once, mesh-wide: each holds one of two -slot keys in `questions`, taken by create or compare-and-set, so two nodes asking at once cannot both take the last, -and a slot that cannot be read refuses the question. Questions alone never fill the router's three open asks, so one -is left for a proposed instruction section while two are open. A question the router holds no record of, its refusal -not heard here, keeps its slot and reads as waiting until it expires: the router may still take it from its queue, +most once a minute, and nothing enforces it. No question is refused for how many wait: the router offers an asker at most three asks at once, holds the oldest back behind newer ones and offers it again when one ends (hq ADR 0666 §12). A question the router holds no record of, its refusal +not heard here, reads as waiting until it expires: the router may still take it from its queue, and no cancel can overtake it there (the queue offers work again after an error, and a cancel of an ask not yet recorded is dropped); once expired, the router refuses it if taken late. A refused question keeps none of its words. An answer is the operator's word to the session and authorises nothing; a question never defaults. diff --git a/modules/claude-code/cmd/claude-code/proposals.go b/modules/claude-code/cmd/claude-code/proposals.go index e3a32a129..3c7198058 100644 --- a/modules/claude-code/cmd/claude-code/proposals.go +++ b/modules/claude-code/cmd/claude-code/proposals.go @@ -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 { diff --git a/modules/claude-code/cmd/claude-code/questions.go b/modules/claude-code/cmd/claude-code/questions.go index 07b5d1231..ea8eb200d 100644 --- a/modules/claude-code/cmd/claude-code/questions.go +++ b/modules/claude-code/cmd/claude-code/questions.go @@ -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 diff --git a/modules/claude-code/cmd/claude-code/questions_test.go b/modules/claude-code/cmd/claude-code/questions_test.go index 8ff3fd74a..a0ffcc8fa 100644 --- a/modules/claude-code/cmd/claude-code/questions_test.go +++ b/modules/claude-code/cmd/claude-code/questions_test.go @@ -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) -}