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

Merge pull request 'claude-code: a session asks the operator on Telegram and reads the answer by a key (issue 589, hq ADR 0332)' (#258) from fix/589-a-session-asks-the-operator into main

Browse Source
This commit was merged in pull request #258.
This commit is contained in:
mesh-admin
2026-10-12 21:30:22 +00:00
parent 9c9a76d220 c4eebce598
commit 43a9155d5a
6 changed files with 1277 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/claude-code/README.md
+23
View File
@@ -68,6 +68,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.<server>` for every node, `<node>.<server>` 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 |
| the agent's configuration (hq ADR 0216) | a key in the module's `config` state per registration — `mesh.<kind>.<name>` for every node, `node.<node>.<kind>.<name>` for one, `home.<node>.<kind>.<name>` 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) |
## The mesh's tools first (hq ADR 0245)
@@ -160,6 +161,28 @@ a secret, more than 16 KiB, a name the terminal sets, and — at once, naming th
that proves its sender is there. The controller's hand-act log records each warrant once (`warranted`), read from the
router's record. The text is never logged.
**A session asks the operator a question on Telegram** (hq ADR 0332, issue 589). An agent that needs a decision while
the operator may be away from the terminal calls `claude_code_ask` with a headline, the context in plain words, two to
four options (a label and what choosing it means, exactly one recommended, with why), what it does if nobody
answers, its name as who@node (required), and an expiry of 5 to 1430 minutes (60 when absent). The ask is published under this
module's name on `operator-channel`; every option is at the level approve and binds the digest of the question and of
itself, so only a channel that proves who answered (Telegram) carries it, never the desk. Anything shaped like a
secret, or a character a phone does not show, in any of its words is refused before anything is kept. The tool waits
a few seconds for the router's word and answers within seconds with the question's key, or the router's refusal (the content rule: no path,
address, host name, commit id or secret's shape; `messenger_check` tests words first). `claude_code_answer` with the
key says waiting, chosen (the option, who answered, on which channel, when), ended without a choice (expired,
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,
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.
The home scope is registered through the tools **only where the home belongs to an agent account of its own**
(hq ADR 0266, issue 340), and only for the node whose module serves the call. Where the agent runs as the operator,
the home is the operator's own `~/.claude`, whose rules and skills load in every session the operator starts, and
modules/claude-code/cmd/claude-code/proposal_tools.go
+97 -1
View File
@@ -6,6 +6,7 @@ import (
"encoding/json"
"errors"
"strings"
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
@@ -20,6 +21,19 @@ type Proposing struct {
Record Recorder
Now func() time.Time
Write WriteManaged
// Questions is the `questions` state (novox/hq ADR 0332); nil is the bus's, made on first use.
Questions QuestionStore
questionsMu sync.Mutex
}
// questions is the `questions` state.
func (d *Proposing) questions() QuestionStore {
d.questionsMu.Lock()
defer d.questionsMu.Unlock()
if d.Questions == nil {
d.Questions = keptQuestions{stdio.State(QuestionsState)}
}
return d.Questions
}
var proposing *Proposing
@@ -60,6 +74,10 @@ func (d *Proposing) watch(p Paths) {
return stdio.Subscribe(asks.Seat+".decided."+Module, func(e stdio.Envelope) error {
var w asks.Warrant
if json.Unmarshal(e.Body, &w) == nil {
if strings.HasPrefix(w.Ask, QuestionPrefix) {
HeardQuestionEnd(d.questions(), w)
return nil
}
if line, err := HeardEnd(p, d.Store, w, d.Now()); err != nil {
say("taking the router's word on %s: %v", w.Ask, err)
} else if line != "" {
@@ -86,7 +104,7 @@ func (d *Proposing) watch(p Paths) {
func proposalTools(p Paths) []stdio.Tool {
notLaunched := errors.New("this module was not launched with the bus; nothing can be proposed")
return []stdio.Tool{
return append([]stdio.Tool{
{Name: "claude_code_instructions_propose",
Description: "Propose an instruction section for every Claude session on the mesh (scope mesh) or on some nodes " +
"(scope node): a rule, a hand-over, the context a later session needs — or, with remove, the removal of one " +
@@ -127,5 +145,83 @@ func proposalTools(p Paths) []stdio.Tool {
proposing.decideNow(p)
return ListProposals(p, proposing.Store, proposing.Bus, strArg(a, "id"), proposing.Now())
}},
}, questionTools(p, notLaunched)...)
}
// questionTools are a session's question to the operator and its answer (novox/hq ADR 0332).
func questionTools(p Paths, notLaunched error) []stdio.Tool {
return []stdio.Tool{
{Name: "claude_code_ask",
Description: "Ask the operator a question on Telegram when you need a decision and the operator may not be at the " +
"terminal (novox/hq ADR 0332): a headline, the context in plain words, two to four options with exactly one " +
"marked recommended and why, what you do if nobody answers, who you are (by, required), and when it expires (5 to " +
"1430 minutes, 60 when absent). Answers within seconds, once the router has taken it, with the question's " +
"key, or with the router's refusal; read the answer with claude_code_answer and keep working " +
"meanwhile. Only a channel that proves who answered carries it, never the desk. The words leave the mesh, so the " +
"router refuses a path, an address, a host name, a commit id or anything shaped like a secret: say it in " +
"words (messenger_check tests words first); anything shaped like a secret is refused before it is kept. At most two " +
"questions wait at once, mesh-wide. The answer is the " +
"operator's word to you; it authorises no call that needs the operator's approval.",
Input: map[string]any{
"headline": str("what is decided, in a few plain words, at most 60 characters, e.g. 'Merge the resolver fix tonight?'"),
"context": str("what the operator decides and why it matters, in plain words, at most 1500 bytes"),
"options": map[string]any{"type": "array", "description": "two to four answers; exactly one recommended",
"items": map[string]any{"type": "object", "properties": map[string]any{
"label": str("the button's words, at most 24 characters"),
"means": str("what you do when it is chosen, in plain words"),
"recommended": map[string]any{"type": "boolean", "description": "true on the one you recommend"},
}, "required": []string{"label", "means"}}},
"why": str("why the recommended option is the one you recommend"),
"on_expiry": str("what you do if nobody answers before it expires, e.g. 'I leave the pull request open'"),
"by": str("required: your session's name, as who@node or who, e.g. fix-589@g14"),
"expires_minutes": map[string]any{"type": "integer",
"description": "minutes until it expires, 5 to 1430; 60 when absent"},
},
Run: func(a map[string]any) (any, error) {
if proposing == nil {
return nil, notLaunched
}
in := QuestionInput{Headline: strArg(a, "headline"), Context: strArg(a, "context"), Why: strArg(a, "why"),
OnExpiry: strArg(a, "on_expiry"), By: strArg(a, "by"), Options: questionOptionsOf(a["options"])}
if m, ok := a["expires_minutes"].(float64); ok {
in.Expires = time.Duration(m * float64(time.Minute))
}
return AskQuestion(p, in, proposing.questions(), proposing.Bus, proposing.Now())
}},
{Name: "claude_code_answer",
Description: "Where a question asked with claude_code_ask stands, by its key (novox/hq ADR 0332): waiting; chosen, " +
"with the option, who answered, on which channel and when; ended without a choice (expired, refused, " +
"cancelled, replaced) with the router's words; or not proven, when the router's record does not hold to the " +
"question asked. Read from the router's own record. wait_seconds (at most 60) waits for an answer within the " +
"call; read a question at most once a minute (asked of you, not enforced).",
Input: map[string]any{
"key": str("the key claude_code_ask gave, e.g. q-0123456789abcdef"),
"wait_seconds": map[string]any{"type": "integer", "description": "wait up to this many seconds (at most 60) while it is waiting"},
},
Run: func(a map[string]any) (any, error) {
if proposing == nil {
return nil, notLaunched
}
wait := time.Duration(0)
if s, ok := a["wait_seconds"].(float64); ok && s > 0 {
wait = time.Duration(s * float64(time.Second))
}
return WaitForAnswer(proposing.questions(), proposing.Bus, strArg(a, "key"), wait, proposing.Now)
}},
}
}
// questionOptionsOf reads the `options` argument: a list of objects with label, means and recommended.
func questionOptionsOf(v any) []QuestionOption {
list, _ := v.([]any)
var out []QuestionOption
for _, item := range list {
m, _ := item.(map[string]any)
if m == nil {
out = append(out, QuestionOption{})
continue
}
out = append(out, QuestionOption{Label: strArg(m, "label"), Means: strArg(m, "means"), Recommended: boolArg(m, "recommended")})
}
return out
}
modules/claude-code/cmd/claude-code/proposals.go
+12 -4
View File
@@ -838,14 +838,20 @@ func HeardEnd(p Paths, store ProposalStore, w asks.Warrant, now time.Time) (stri
if !known || (w.About != "" && w.About != prop.Ask().About) {
return "", nil
}
endedMu.Lock()
ended[w.Ask] = w
endedMu.Unlock()
// Said to a proposer waiting on it only once the discard is kept: what it reads next is already there (the
// flake of the router's refusal test under a loaded build agent, 2026-10-12).
tell := func() {
endedMu.Lock()
ended[w.Ask] = w
endedMu.Unlock()
}
approved, err := ReadApproved(p)
if err != nil {
tell()
return "", err
}
if _, done := approved.Decided[w.Ask]; done {
tell()
return "", nil
}
// Heard, not read from a record: Reconcile still reads the router's record of it, so an ask the router took after
@@ -856,7 +862,9 @@ func HeardEnd(p Paths, store ProposalStore, w asks.Warrant, now time.Time) (stri
}
approved.Decided[w.Ask] = d
delete(approved.Refused, w.Ask)
if err := writeApproved(p, approved); err != nil {
err = writeApproved(p, approved)
tell()
if err != nil {
return "", err
}
return "proposal " + w.Ask + ": " + d.Outcome, nil
modules/claude-code/cmd/claude-code/questions.go
+628
View File
@@ -0,0 +1,628 @@
package main
// A question an agent session asks the operator, answered on a channel that proves who answered (novox/hq ADR 0332,
// issue 589).
//
// A session that needs a yes, a no or a choice asks it with claude_code_ask instead of waiting at a terminal the
// operator may not be watching. The ask is published under this module's name on the operator channel (ADR 0259):
// the bus grants that subject to this module alone, so the asker is proven by the bus, and the router stays the judge
// of answers, never an asker for others. Every option is at the level approve, so only a channel that proves who
// answered (Telegram, today) carries it, and each option binds the digest of the question and of itself.
//
// The session reads the answer by the question's key with claude_code_answer: from the router's own record of the
// ask, which only the router writes and only this module reads, held to the question as the `questions` state keeps
// it (asks.Warrant.For and asks.Option.Performs). A record that does not hold is said as not proven, never as a
// choice. An answer informs the session; it authorises nothing (ADR 0286 stays the way to an approved call).
import (
"crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"regexp"
"sort"
"strconv"
"strings"
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/asks"
)
// QuestionsState is the module's state the questions are kept in.
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
// (asks.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 = asks.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 (asks.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"`
}
// 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.
const (
QuestionShortest = 5 * time.Minute
QuestionLongest = ProposalLasts
QuestionDefault = time.Hour
)
// The bounds of what a question says.
const (
FewestQuestionOptions = 2
MostContextBytes = 1500
MostMeansBytes = 200
MostWhyBytes = 300
MostOnExpiryBytes = 200
)
// How long one claude_code_answer call may wait for the answer, and how often it looks meanwhile.
var (
AnswerWaitsAtMost = 60 * time.Second
AnswerLooksEvery = 2 * time.Second
)
// QuestionOption is one answer a session offers.
type QuestionOption struct {
ID string `json:"id"`
Label string `json:"label"`
Means string `json:"means"`
Recommended bool `json:"recommended,omitempty"`
}
// QuestionStore is the `questions` state: a value with its revision, created only when absent, updated only at the
// revision read.
type QuestionStore interface {
Get(key string) (json.RawMessage, uint64, bool, error)
Create(key string, value any) error
Update(key string, value any, revision uint64) error
Put(key string, value any) error
}
// keptQuestions is the `questions` state on the bus.
type keptQuestions struct{ s stdio.KeptState }
func (k keptQuestions) Get(key string) (json.RawMessage, uint64, bool, error) {
e, err := k.s.Get(key)
if err != nil || e == nil {
return nil, 0, false, err
}
return e.Value, e.Revision, true, nil
}
func (k keptQuestions) Create(key string, value any) error {
_, err := k.s.Create(key, value)
return err
}
func (k keptQuestions) Update(key string, value any, revision uint64) error {
_, err := k.s.Update(key, value, revision)
return err
}
func (k keptQuestions) Put(key string, value any) error { _, err := k.s.Put(key, value); return err }
// Question is one question, as the `questions` state keeps it. Every field the ask is composed from is here, so the
// answer is held to exactly the question asked.
type Question struct {
ID string `json:"id"`
Headline string `json:"headline"`
Context string `json:"context"`
Options []QuestionOption `json:"options"`
Why string `json:"why"`
OnExpiry string `json:"on-expiry"`
By string `json:"by"`
From string `json:"from"`
At time.Time `json:"at"`
Expires time.Time `json:"expires"`
// Asked is empty once the ask was published, or says why it was not; nothing composes from it.
Asked string `json:"asked,omitempty"`
// Ended is the router's word that the ask ended without a choice, as heard on the decided subject: a refused ask
// leaves no record of the router's to read it from. It never makes a choice.
Ended string `json:"ended,omitempty"`
}
// recommended is the option marked as recommended.
func (q Question) recommended() (QuestionOption, bool) {
for _, o := range q.Options {
if o.Recommended {
return o, true
}
}
return QuestionOption{}, false
}
// Act is what an answer binds: this question, whole, and the option chosen.
func (q Question) Act(option string) asks.Act {
act := asks.Act{"verb": Module + ".question", "option": option, "question": q.ID, "headline": q.Headline,
"context": q.Context, "why": q.Why, "on-expiry": q.OnExpiry, "by": q.By, "from": q.From,
"at": q.At.UTC().Format(time.RFC3339Nano), "expires": q.Expires.UTC().Format(time.RFC3339Nano),
"options": strconv.Itoa(len(q.Options))}
for i, o := range q.Options {
act[fmt.Sprintf("option.%d", i)] = o.ID + "\n" + o.Label + "\n" + o.Means + "\n" + strconv.FormatBool(o.Recommended)
}
return act
}
// Ask is the ask for a question, composed from it alone.
func (q Question) Ask() asks.Ask {
var b strings.Builder
b.WriteString(strings.TrimSpace(q.Context))
if rec, ok := q.recommended(); ok {
fmt.Fprintf(&b, "\n\nRecommended: %s, because %s", rec.Label, strings.TrimRight(strings.TrimSpace(q.Why), "."))
b.WriteString(".")
}
fmt.Fprintf(&b, "\n\nAsked by the agent session %s on %s through the claude-code module, at %s UTC. "+
"Your answer is told to that session; it performs nothing by itself.", q.By, q.From, q.At.UTC().Format("15:04 on 2 Jan"))
a := asks.Ask{ID: q.ID, Headline: q.Headline, Explanation: b.String(), Who: asks.Operator, Expires: q.Expires,
OnExpiry: q.OnExpiry}
for _, o := range q.Options {
does := o.Means
if o.Recommended {
does = "Recommended. " + does
}
binds, _ := asks.ActDigest(q.Act(o.ID))
a.Options = append(a.Options, asks.Option{ID: o.ID, Label: o.Label, Does: does, Level: asks.Approve, Binds: binds})
}
return a
}
var sessionName = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_-]{0,47}$`)
// QuestionProblem says why a question cannot be asked, or "": every bound of ADR 0332 §2 that this module holds
// itself. The words are held to the content rule by the router, which says its refusal back.
func QuestionProblem(q Question) string {
var problems []string
add := func(format string, args ...any) { problems = append(problems, fmt.Sprintf(format, args...)) }
switch {
case strings.TrimSpace(q.Headline) == "":
add("it has no headline")
case len([]rune(q.Headline)) > asks.HeadlineLength:
add("its headline is %d characters, more than %d", len([]rune(q.Headline)), asks.HeadlineLength)
}
switch {
case strings.TrimSpace(q.Context) == "":
add("it gives no context: say in plain words what the operator decides and why it matters")
case len(q.Context) > MostContextBytes:
add("its context is %d bytes, more than the %d a phone shows in one glance", len(q.Context), MostContextBytes)
}
if n := len(q.Options); n < FewestQuestionOptions || n > asks.MostOptions {
add("it offers %d options; a question offers %d to %d", n, FewestQuestionOptions, asks.MostOptions)
}
recommended, labels := 0, map[string]bool{}
for i, o := range q.Options {
switch {
case strings.TrimSpace(o.Label) == "":
add("option %d has no label", i+1)
case len([]rune(o.Label)) > asks.LabelLength:
add("option %d's label is longer than %d characters", i+1, asks.LabelLength)
case labels[strings.ToLower(o.Label)]:
add("two options are labelled %q", o.Label)
}
labels[strings.ToLower(o.Label)] = true
if strings.TrimSpace(o.Means) == "" {
add("option %d does not say what choosing it means", i+1)
} else if len(o.Means) > MostMeansBytes {
add("what option %d means is longer than %d bytes", i+1, MostMeansBytes)
}
if !asks.UsableID(o.ID) {
add("option %d's id is not one", i+1)
}
if o.Recommended {
recommended++
}
}
if recommended != 1 {
add("%d options are marked recommended; exactly one is", recommended)
}
switch {
case strings.TrimSpace(q.Why) == "":
add("it does not say why the recommended option is recommended")
case len(q.Why) > MostWhyBytes:
add("why the option is recommended is longer than %d bytes", MostWhyBytes)
}
switch {
case strings.TrimSpace(q.OnExpiry) == "":
add("it does not say what the session does if nobody answers")
case len(q.OnExpiry) > MostOnExpiryBytes:
add("what happens if nobody answers is longer than %d bytes", MostOnExpiryBytes)
}
if !sessionName.MatchString(q.By) {
add("who asks is not a session's name: letters, digits, - and _, such as fix-589")
}
if !nodeName.MatchString(q.From) {
add("where it comes from is not a node's name")
}
if life := q.Expires.Sub(q.At); life < QuestionShortest || life > QuestionLongest {
add("it lasts %s; a question lasts %s to %s", life.Round(time.Minute), QuestionShortest, QuestionLongest)
}
if problem := questionWordsProblem(q); problem != "" {
add("%s", problem)
}
if len(problems) == 0 {
return ""
}
return strings.Join(problems, "; ")
}
// questionWordsProblem says why a question's words may not be kept at all, or "": anything shaped like a secret, or a
// character a phone does not show, in any word the operator is shown — the headline, the context, why, what happens
// on expiry, and each option's label and meaning. Judged before anything is kept: the `questions` state is read by
// every node, and the router judges the content rule only after the question was kept.
func questionWordsProblem(q Question) string {
fields := []struct{ name, text string }{{"the headline", q.Headline}, {"the context", q.Context}, {"why", q.Why},
{"what happens on expiry", q.OnExpiry}}
for i, o := range q.Options {
fields = append(fields, struct{ name, text string }{fmt.Sprintf("option %d's label", i+1), o.Label},
struct{ name, text string }{fmt.Sprintf("what option %d means", i+1), o.Means})
}
for _, f := range fields {
for _, re := range secretShaped {
if re.MatchString(f.text) {
return f.name + " holds something shaped like a secret, which is never kept or shown on a phone"
}
}
if what := invisible(f.text); what != "" {
return f.name + " holds " + what + ", which a phone does not show"
}
}
return ""
}
func newQuestionID() string {
var b [8]byte
_, _ = rand.Read(b[:])
return QuestionPrefix + hex.EncodeToString(b[:])
}
// QuestionInput is what a session asks.
type QuestionInput struct {
Headline string
Context string
Options []QuestionOption
Why string
OnExpiry string
By string
Expires time.Duration
}
// AskQuestion keeps a question and asks the operator. It answers within seconds, once the router has said whether it
// took the question, with the question's key; the answer is read with ReadAnswer.
func AskQuestion(p Paths, in QuestionInput, store QuestionStore, bus AskBus, now time.Time) (map[string]any, error) {
refuse := func(why string) (map[string]any, error) { return nil, errors.New(why + ". Nothing was asked") }
by := strings.TrimSpace(in.By)
if who, node, found := strings.Cut(by, "@"); found {
if node != p.Node {
return refuse(fmt.Sprintf("who asks names the node %q, and this is %s: ask on the node the session runs on", node, p.Node))
}
by = who
}
if by == "" {
return refuse("it does not say who asks: give by, your session's name as who@node, such as fix-589@g14")
}
life := in.Expires
if life == 0 {
life = QuestionDefault
}
q := Question{ID: newQuestionID(), Headline: strings.TrimSpace(in.Headline), Context: strings.TrimSpace(in.Context),
Why: strings.TrimSpace(in.Why), OnExpiry: strings.TrimSpace(in.OnExpiry), By: by, From: p.Node, At: now.UTC(),
Expires: now.UTC().Add(life)}
for i, o := range in.Options {
q.Options = append(q.Options, QuestionOption{ID: fmt.Sprintf("o%d", i+1), Label: strings.TrimSpace(o.Label),
Means: strings.TrimSpace(o.Means), Recommended: o.Recommended})
}
if problem := QuestionProblem(q); problem != "" {
return refuse(problem)
}
ask := q.Ask()
if err := ask.Check(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(asks.AskSubject(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)
}
say("question %s from the session %s asked of the operator", q.ID, q.By)
taken := "the router has not said yet whether it took the question; claude_code_answer says"
defer forgetQuestionEnd(q.ID)
for deadline := time.Now().Add(RouterAnswersWithin); time.Now().Before(deadline); {
if w, ended := questionEnd(q.ID); ended {
return nil, fmt.Errorf("the router did not ask the operator: the question %s (%s)", w.Outcome, w.Words)
}
if entry, err := bus.Record(q.ID); err == nil && entry != nil {
taken = "the router took the question and shows it where who answers is proven (Telegram)"
break
}
time.Sleep(RouterAnswersEvery)
}
return map[string]any{
"key": q.ID,
"router": taken,
"asked": "\"" + q.Headline + "\", with the answers " + labelsOf(q),
"expires": q.Expires.Format(time.RFC3339),
"next": "read the answer with claude_code_answer {\"key\": \"" + q.ID + "\"} (it may wait up to a minute); " +
"keep working meanwhile, and if it expires do what you said: " + q.OnExpiry,
}, nil
}
func labelsOf(q Question) string {
var out []string
for _, o := range q.Options {
out = append(out, o.Label)
}
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 {
return Question{ID: q.ID, By: q.By, From: q.From, At: q.At, Expires: q.Expires, Ended: ended}
}
func readQuestion(store QuestionStore, key string) (Question, bool) {
q, ok, _ := readQuestionErr(store, key)
return q, ok
}
func readQuestionErr(store QuestionStore, key string) (Question, bool, error) {
if !strings.HasPrefix(key, QuestionPrefix) {
return Question{}, false, nil
}
raw, _, found, err := store.Get(key)
if err != nil || !found {
return Question{}, false, err
}
var q Question
if json.Unmarshal(raw, &q) != nil || q.ID != key {
return Question{}, false, nil
}
return q, true, nil
}
// The router's word that a question ended without a choice, as heard on the decided subject.
var (
questionEndsMu sync.Mutex
questionEnds = map[string]asks.Warrant{}
)
func questionEnd(id string) (asks.Warrant, bool) {
questionEndsMu.Lock()
defer questionEndsMu.Unlock()
w, ok := questionEnds[id]
return w, ok
}
func forgetQuestionEnd(id string) {
questionEndsMu.Lock()
delete(questionEnds, id)
questionEndsMu.Unlock()
}
// HeardQuestionEnd takes the router's word that a question ended without a choice — refused, expired, cancelled or
// replaced — and keeps it on the question, because a refused ask leaves no record of the router's. A refused question
// keeps none of its words (tombstone). A choice is never taken from this word: only from the router's record, held to
// the question (ReadAnswer).
func HeardQuestionEnd(store QuestionStore, w asks.Warrant) {
if w.Asker != Module || w.Outcome == asks.OutcomeChosen || !strings.HasPrefix(w.Ask, QuestionPrefix) {
return
}
q, known := readQuestion(store, w.Ask)
if !known {
return
}
if q.Ended == "" {
ended := string(w.Outcome)
if w.Words != "" {
ended += ": " + w.Words
}
if w.Outcome == asks.OutcomeRefused {
q = tombstone(q, ended)
} else {
q.Ended = ended
}
_ = store.Put(q.ID, q)
}
// Kept on the question first, then said to a session waiting on it: what it reads next is already there.
questionEndsMu.Lock()
questionEnds[w.Ask] = w
questionEndsMu.Unlock()
}
// The answer's states, as claude_code_answer says them.
const (
AnswerWaiting = "waiting"
AnswerChosen = "chosen"
AnswerEnded = "ended"
AnswerNotProven = "not proven"
)
// ReadAnswer is where a question stands: waiting, chosen (and what), ended without a choice, or not proven.
func ReadAnswer(store QuestionStore, bus AskBus, key string, now time.Time) (map[string]any, error) {
q, ok := readQuestion(store, strings.TrimSpace(key))
if !ok {
return nil, fmt.Errorf("no question %q is kept: the key is the one claude_code_ask gave", key)
}
out := map[string]any{"key": q.ID, "headline": q.Headline, "asked": q.At.Format(time.RFC3339),
"expires": q.Expires.Format(time.RFC3339)}
if q.Asked != "" {
out["state"], out["words"] = AnswerEnded, q.Asked
return out, nil
}
entry, err := bus.Record(q.ID)
if err != nil {
return nil, fmt.Errorf("the router's record of the question could not be read now (%v); ask again", err)
}
var rec routerRecord
if entry != nil {
if err := json.Unmarshal(entry.Value, &rec); err != nil {
out["state"], out["words"] = AnswerNotProven, "the router's record of it cannot be read: "+err.Error()
return out, nil
}
}
switch {
case entry == nil && q.Ended != "":
out["state"], out["words"] = AnswerEnded, q.Ended
case entry == nil && !now.Before(q.Expires):
out["state"], out["words"] = AnswerEnded, "expired without the router recording it: refused (its refusal was not "+
"heard here) or never taken; the router refuses it now if it is taken late"
out["now"] = "do what you said: " + q.OnExpiry
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 " +
q.Expires.Format(time.RFC3339)
case entry == nil, rec.State == "open", rec.Warrant == nil:
out["state"] = AnswerWaiting
if q.Ended != "" {
out["words"] = "the router said it ended (" + q.Ended + "), and its record says it is open; read again"
}
default:
w := *rec.Warrant
if w.Outcome != asks.OutcomeChosen {
out["state"], out["words"] = AnswerEnded, string(w.Outcome)
if w.Words != "" {
out["words"] = string(w.Outcome) + ": " + w.Words
}
if w.Outcome == asks.OutcomeExpired {
out["now"] = "do what you said: " + q.OnExpiry
}
return out, nil
}
if problem := QuestionProblem(q); problem != "" {
out["state"], out["words"] = AnswerNotProven, "the question kept is not one this module asks: "+problem
return out, nil
}
option, err := w.For(Module, q.Ask())
if err == nil {
err = option.Performs(q.Act(option.ID))
}
if err != nil {
say("REFUSED the answer to question %s: %v", q.ID, err)
out["state"], out["words"] = AnswerNotProven, "the router's record does not hold to the question asked: "+err.Error()
return out, nil
}
chosen := map[string]any{"id": option.ID, "label": option.Label}
for _, o := range q.Options {
if o.ID == option.ID {
chosen["means"], chosen["recommended"] = o.Means, o.Recommended
}
}
out["state"], out["chosen"], out["by"], out["channel"], out["at"] = AnswerChosen, chosen, w.Says(), w.Channel,
w.At.Format(time.RFC3339)
out["note"] = "the operator's word to this session; it authorises no call that needs the operator's approval"
}
return out, nil
}
// WaitForAnswer reads the answer, and while it is waiting reads it again until wait has passed.
func WaitForAnswer(store QuestionStore, bus AskBus, key string, wait time.Duration, now func() time.Time) (map[string]any, error) {
if wait > AnswerWaitsAtMost {
wait = AnswerWaitsAtMost
}
deadline := time.Now().Add(wait)
for {
out, err := ReadAnswer(store, bus, key, now())
if err != nil || out["state"] != AnswerWaiting || !time.Now().Before(deadline) {
return out, err
}
time.Sleep(AnswerLooksEvery)
}
}
modules/claude-code/cmd/claude-code/questions_test.go
+513
View File
@@ -0,0 +1,513 @@
package main
// A session's question to the operator (novox/hq ADR 0332, issue 589): asked under this module's name at the level
// approve, refused outside its bounds, at most two waiting, and its answer read back by its key only when the
// router's record holds to the question asked.
import (
"encoding/json"
"errors"
"strings"
"sync"
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/asks"
)
func mergeTonight() QuestionInput {
return QuestionInput{
Headline: "Merge the resolver fix tonight?",
Context: "The fix for the resolver on the laptop is reviewed and both checks are green. Merging now rolls it to every node within the hour.",
Options: []QuestionOption{
{Label: "Merge now", Means: "I merge it and follow the rollout", Recommended: true},
{Label: "Wait", Means: "I leave the pull request open until you say"},
},
Why: "it is reviewed and the checks are green",
OnExpiry: "I leave the pull request open",
By: "fix-589@g14",
}
}
func askMergeTonight(t *testing.T, p Paths, store *memQuestions, r *router, at time.Time) string {
t.Helper()
out, err := AskQuestion(p, mergeTonight(), store, r, at)
if err != nil {
t.Fatal(err)
}
return out["key"].(string)
}
func TestAQuestionIsAskedUnderTheModulesNameAtTheLevelApprove(t *testing.T) {
p, _ := node(t, "g14")
store, r := newMemQuestions(), newRouter()
out, err := AskQuestion(p, mergeTonight(), store, r, t0)
if err != nil {
t.Fatal(err)
}
key := out["key"].(string)
if !strings.HasPrefix(key, QuestionPrefix) || !strings.Contains(out["router"].(string), "took the question") {
t.Fatalf("answered %v", out)
}
if len(r.subjects) != 1 || r.subjects[0] != asks.AskSubject(Module) {
t.Fatalf("published on %v", r.subjects)
}
a := r.published[key]
if a.Who != asks.Operator || a.OnExpiry != "I leave the pull request open" || !a.Expires.Equal(t0.Add(QuestionDefault)) {
t.Errorf("ask %+v", a)
}
if len(a.Options) != 2 {
t.Fatalf("options %+v", a.Options)
}
for _, o := range a.Options {
if o.Level != asks.Approve || o.Binds == "" {
t.Errorf("option %s is %s binding %q: every option is an approval bound to the question", o.ID, o.Level, o.Binds)
}
}
if !strings.HasPrefix(a.Options[0].Does, "Recommended. ") || strings.HasPrefix(a.Options[1].Does, "Recommended") {
t.Errorf("the recommendation is not on the recommended option alone: %+v", a.Options)
}
for _, want := range []string{"Recommended: Merge now, because it is reviewed and the checks are green.",
"the agent session fix-589 on g14", "performs nothing by itself"} {
if !strings.Contains(a.Explanation, want) {
t.Errorf("the explanation does not say %q:\n%s", want, a.Explanation)
}
}
if err := a.Check(t0); err != nil {
t.Errorf("the ask does not pass the asks contract: %v", err)
}
}
func TestTheChosenAnswerIsReadBackByItsKey(t *testing.T) {
p, _ := node(t, "g14")
store, r := newMemQuestions(), newRouter()
key := askMergeTonight(t, p, store, r, t0)
out, err := ReadAnswer(store, r, key, t0.Add(time.Minute))
if err != nil || out["state"] != AnswerWaiting {
t.Fatalf("before an answer: %v %v", out, err)
}
r.close(r.chose(key, "o2", t0.Add(10*time.Minute)))
out, err = ReadAnswer(store, r, key, t0.Add(11*time.Minute))
if err != nil || out["state"] != AnswerChosen {
t.Fatalf("after the choice: %v %v", out, err)
}
chosen := out["chosen"].(map[string]any)
if chosen["id"] != "o2" || chosen["label"] != "Wait" || chosen["means"] != "I leave the pull request open until you say" {
t.Errorf("chosen %v", chosen)
}
if !strings.Contains(out["by"].(string), "the operator, via telegram") || out["channel"] != "telegram" {
t.Errorf("by %v on %v", out["by"], out["channel"])
}
}
func TestAnAnswerThatDoesNotHoldToTheQuestionIsNotProven(t *testing.T) {
p, _ := node(t, "g14")
store, r := newMemQuestions(), newRouter()
key := askMergeTonight(t, p, store, r, t0)
w := r.chose(key, "o1", t0.Add(time.Minute))
// The question rewritten on the bus after it was asked: the operator was shown other words.
var q Question
_ = json.Unmarshal(store.m[key], &q)
q.Context = "Merging now deletes nothing and rolls it to every node."
_ = store.Put(key, q)
r.close(w)
if out, _ := ReadAnswer(store, r, key, t0.Add(2*time.Minute)); out["state"] != AnswerNotProven {
t.Errorf("a question changed after it was asked read as %v", out)
}
// A warrant for another ask's digest.
store2, r2 := newMemQuestions(), newRouter()
key2 := askMergeTonight(t, p, store2, r2, t0)
w2 := r2.chose(key2, "o1", t0.Add(time.Minute))
w2.AskDigest = "sha256:0000"
r2.close(w2)
if out, _ := ReadAnswer(store2, r2, key2, t0.Add(2*time.Minute)); out["state"] != AnswerNotProven {
t.Errorf("a warrant for another ask read as %v", out)
}
// A choice made after the question expired.
store3, r3 := newMemQuestions(), newRouter()
key3 := askMergeTonight(t, p, store3, r3, t0)
r3.close(r3.chose(key3, "o1", t0.Add(QuestionDefault+time.Minute)))
if out, _ := ReadAnswer(store3, r3, key3, t0.Add(2*QuestionDefault)); out["state"] != AnswerNotProven {
t.Errorf("a choice after expiry read as %v", out)
}
}
func TestAQuestionIsRefusedOutsideItsBounds(t *testing.T) {
p, _ := node(t, "g14")
for name, change := range map[string]func(*QuestionInput){
"one option": func(in *QuestionInput) { in.Options = in.Options[:1] },
"five options": func(in *QuestionInput) { in.Options = append(in.Options, in.Options[1], in.Options[1], in.Options[1]) },
"none recommended": func(in *QuestionInput) { in.Options[0].Recommended = false },
"two recommended": func(in *QuestionInput) { in.Options[1].Recommended = true },
"no why": func(in *QuestionInput) { in.Why = "" },
"no context": func(in *QuestionInput) { in.Context = "" },
"no on-expiry": func(in *QuestionInput) { in.OnExpiry = "" },
"a long headline": func(in *QuestionInput) { in.Headline = strings.Repeat("Merge ", 11) },
"a long label": func(in *QuestionInput) { in.Options[0].Label = "Merge it now and follow the rollout" },
"an option unsaid": func(in *QuestionInput) { in.Options[1].Means = "" },
"too short": func(in *QuestionInput) { in.Expires = time.Minute },
"too long": func(in *QuestionInput) { in.Expires = 2 * 24 * time.Hour },
"another node": func(in *QuestionInput) { in.By = "fix-589@shanks" },
"an invisible char": func(in *QuestionInput) { in.Context += "\u200b" },
"invisible label": func(in *QuestionInput) { in.Options[0].Label = "Merge\u200bnow" },
"invisible means": func(in *QuestionInput) { in.Options[1].Means += "\u202e" },
"a token, context": func(in *QuestionInput) { in.Context += " The key is ghp_abcdefghijklmnopqrstuvwxyz0123456789." },
"a token, headline": func(in *QuestionInput) { in.Headline = "Use sk-abcdefghijklmnopqrstuv?" },
"a token, why": func(in *QuestionInput) { in.Why = "password: hunter2 works" },
"a token, expiry": func(in *QuestionInput) { in.OnExpiry = "I keep AKIAABCDEFGHIJKLMNOP" },
"a token, label": func(in *QuestionInput) { in.Options[0].Label = "token=abc" },
"a token, means": func(in *QuestionInput) { in.Options[1].Means = "-----BEGIN PRIVATE KEY-----" },
"no by": func(in *QuestionInput) { in.By = "" },
"61 characters": func(in *QuestionInput) { in.Headline = strings.Repeat("x", 60) + "?" },
"1431 minutes": func(in *QuestionInput) { in.Expires = 1431 * time.Minute },
} {
store, r := newMemQuestions(), newRouter()
in := mergeTonight()
in.Options = append([]QuestionOption(nil), in.Options...)
change(&in)
if _, err := AskQuestion(p, in, store, r, t0); err == nil || !strings.Contains(err.Error(), "Nothing was asked") {
t.Errorf("%s: %v", name, err)
}
if len(r.subjects) != 0 || len(store.m) != 0 {
t.Errorf("%s: asked or kept anyway", name)
}
}
}
func TestAtMostTwoQuestionsWaitSoAProposalFindsRoom(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)
}
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)
}
// 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)
}
}
// questionRefusingRouter refuses every ask as the router does for words that may not leave the mesh: on the decided
// subject, with no record.
type questionRefusingRouter struct {
*router
store QuestionStore
id string
}
func (r *questionRefusingRouter) Publish(subject string, body any, id string) error {
r.id = id
go func() {
time.Sleep(50 * time.Millisecond)
HeardQuestionEnd(r.store, asks.Warrant{Ask: id, Asker: Module, Outcome: asks.OutcomeRefused,
Words: "its words carry path (a file path), which may not leave the mesh", At: t0})
}()
return nil
}
func TestTheRoutersRefusalIsSaidAndKept(t *testing.T) {
p, _ := node(t, "g14")
store := newMemQuestions()
refusing := &questionRefusingRouter{router: newRouter(), store: store}
within, every := RouterAnswersWithin, RouterAnswersEvery
RouterAnswersWithin, RouterAnswersEvery = time.Second, 10*time.Millisecond
t.Cleanup(func() { RouterAnswersWithin, RouterAnswersEvery = within, every })
_, err := AskQuestion(p, mergeTonight(), store, refusing, t0)
if err == nil || !strings.Contains(err.Error(), "may not leave the mesh") {
t.Fatalf("refusal: %v", err)
}
out, err := ReadAnswer(store, refusing, refusing.id, t0.Add(time.Minute))
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 != "" {
t.Errorf("a refused question keeps its words: %+v", kept)
}
}
func TestAnExpiredQuestionSaysWhatTheSessionDoes(t *testing.T) {
p, _ := node(t, "g14")
store, r := newMemQuestions(), newRouter()
key := askMergeTonight(t, p, store, r, t0)
r.close(asks.Warrant{Ask: key, Outcome: asks.OutcomeExpired, At: t0.Add(QuestionDefault)})
out, err := ReadAnswer(store, r, key, t0.Add(QuestionDefault+time.Minute))
if err != nil || out["state"] != AnswerEnded || out["now"] != "do what you said: I leave the pull request open" {
t.Errorf("expired: %v %v", out, err)
}
}
func TestTheQuestionToolsAreServed(t *testing.T) {
p, _ := node(t, "g14")
r := newRouter()
questions := newMemQuestions()
proposing = &Proposing{Store: &memProposals{m: map[string]json.RawMessage{}}, Bus: r, Now: func() time.Time { return t0 },
Write: writer(map[string]string{}), Questions: questions}
t.Cleanup(func() { proposing = nil })
s := serve(t, p, &bus{kept: map[string]map[string]any{}}, memConfig{})
text, isErr := s.call(t, "claude_code_ask", map[string]any{
"headline": "Merge the resolver fix tonight?", "context": "Reviewed and green.",
"options": []any{
map[string]any{"label": "Merge now", "means": "I merge it", "recommended": true},
map[string]any{"label": "Wait", "means": "I leave it open"},
},
"why": "it is reviewed", "on_expiry": "I leave it open", "by": "fix-589@g14", "expires_minutes": 30,
})
if isErr || !strings.Contains(text, `"key"`) {
t.Fatalf("%v %s", isErr, text)
}
var asked map[string]any
_ = json.Unmarshal([]byte(text), &asked)
key, _ := asked["key"].(string)
if a := r.published[key]; !a.Expires.Equal(t0.Add(30 * time.Minute)) {
t.Errorf("expires %v", a.Expires)
}
r.close(r.chose(key, "o1", t0.Add(time.Minute)))
text, isErr = s.call(t, "claude_code_answer", map[string]any{"key": key})
if isErr || !strings.Contains(text, `"chosen"`) || !strings.Contains(text, "Merge now") {
t.Fatalf("%v %s", isErr, text)
}
}
// One read waits at most AnswerWaitsAtMost (60 s), whatever the session asks for, and says waiting when nobody
// answered meanwhile.
func TestAReadWaitsNoLongerThanItsBound(t *testing.T) {
if AnswerWaitsAtMost != 60*time.Second {
t.Fatalf("a read may wait %s; ADR 0332 says 60 s", AnswerWaitsAtMost)
}
p, _ := node(t, "g14")
store, r := newMemQuestions(), newRouter()
key := askMergeTonight(t, p, store, r, t0)
most, every := AnswerWaitsAtMost, AnswerLooksEvery
AnswerWaitsAtMost, AnswerLooksEvery = 100*time.Millisecond, 10*time.Millisecond
t.Cleanup(func() { AnswerWaitsAtMost, AnswerLooksEvery = most, every })
start := time.Now()
out, err := WaitForAnswer(store, r, key, time.Hour, func() time.Time { return t0.Add(time.Minute) })
if err != nil || out["state"] != AnswerWaiting {
t.Fatalf("%v %v", out, err)
}
if took := time.Since(start); took > 2*time.Second {
t.Errorf("a read asked to wait an hour took %s", took)
}
}
// memQuestions is the `questions` state: values with revisions, create-only and compare-and-set, safe for the
// goroutines a test starts.
type memQuestions struct {
mu sync.Mutex
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 {
return &memQuestions{m: map[string]json.RawMessage{}, rev: map[string]uint64{}}
}
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")
}
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
}
func (s *memQuestions) put(key string, value any) error {
raw, err := json.Marshal(value)
s.n++
s.m[key], s.rev[key] = raw, s.n
return err
}
func (s *memQuestions) Create(key string, value any) error {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.m[key]; ok {
return stdio.ErrChanged
}
return s.put(key, value)
}
func (s *memQuestions) Update(key string, value any, revision uint64) error {
s.mu.Lock()
defer s.mu.Unlock()
if s.rev[key] != revision {
return stdio.ErrChanged
}
return s.put(key, value)
}
func (s *memQuestions) Put(key string, value any) error {
s.mu.Lock()
defer s.mu.Unlock()
return s.put(key, value)
}
// The edges of ADR 0332 §2 that are allowed: a 60-character headline and 1430 minutes.
func TestTheEdgesOfAQuestionAreAsked(t *testing.T) {
p, _ := node(t, "g14")
in := mergeTonight()
in.Headline = strings.Repeat("x", 59) + "?"
in.Expires = 1430 * time.Minute
if _, err := AskQuestion(p, in, newMemQuestions(), newRouter(), t0); err != nil {
t.Fatalf("a 60-character headline lasting 1430 minutes: %v", err)
}
}
// A warrant naming another asker or another ask, or an option the question did not offer, is no answer to it.
func TestAWarrantForAnotherAskOrOptionIsNotProven(t *testing.T) {
p, _ := node(t, "g14")
for name, change := range map[string]func(*asks.Warrant){
"another asker": func(w *asks.Warrant) { w.Asker = "mesh-controller" },
"another ask": func(w *asks.Warrant) { w.Ask = "q-0000000000000000" },
"an option unoffered": func(w *asks.Warrant) { w.Option, w.Label = "o9", "Delete everything" },
} {
store, r := newMemQuestions(), newRouter()
key := askMergeTonight(t, p, store, r, t0)
w := r.chose(key, "o1", t0.Add(time.Minute))
change(&w)
r.records[key], _ = json.Marshal(map[string]any{"state": "chosen", "warrant": w})
if out, _ := ReadAnswer(store, r, key, t0.Add(2*time.Minute)); out["state"] != AnswerNotProven {
t.Errorf("%s read as %v", name, out)
}
}
}
// 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(asks.Warrant{Ask: id, Outcome: asks.OutcomeExpired, At: 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(asks.AskSubject(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)
}
modules/claude-code/module.json
+4 -1
View File
@@ -18,7 +18,8 @@
"servers",
"holdings",
"config",
"proposals"
"proposals",
"questions"
],
"reads": [
"claude-licence-manager.bindings"
@@ -54,6 +55,8 @@
"claude_code_instructions_unregister",
"claude_code_instructions_propose",
"claude_code_proposals",
"claude_code_ask",
"claude_code_answer",
"claude_code_settings_get",
"claude_code_settings_set",
"claude_code_settings_clear",