Keep no secret-shaped question, and hold the two-question cap across nodes (issue 589, review 105)
mesh/merge-gate head 14999f1d3ecb merged into trunk 9c9a76d220f8 — pass: builds claude-code → ace, g14, novox, shanks; no bus step; every machine composes …
mesh/repo-check head 14999f1d3ecb merged into trunk 9c9a76d220f8 — pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-…
mesh/delivery superseded: a newer head of the same pull request
mesh/merge-gate head 14999f1d3ecb merged into trunk 9c9a76d220f8 — pass: builds claude-code → ace, g14, novox, shanks; no bus step; every machine composes …
mesh/repo-check head 14999f1d3ecb merged into trunk 9c9a76d220f8 — pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-…
mesh/delivery superseded: a newer head of the same pull request
A question was kept on the bus before the router judged its words, so a token in it reached a bucket every node reads: every word, the options' too, is now held to the secret shapes and invisible characters before anything is kept, and a refused question keeps none of its words. The cap was check-then-create and failed open: each open question now holds one of two slot keys taken by create or compare-and-set, refusing when a slot cannot be read. A question the router never recorded frees its slot after two minutes instead of at expiry. by is required; wording matches ADR 0332.
This commit is contained in:
@@ -68,7 +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 |
|
||||
| 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)
|
||||
@@ -164,17 +164,22 @@ 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, and an expiry of 5 to 1430 minutes (60 when absent). The ask is published under this
|
||||
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. The tool waits a few seconds
|
||||
for the router's word and answers with the question's key, or the router's refusal (the content rule: no path,
|
||||
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; a session reads a question at most once a
|
||||
minute. At most two questions wait at once, mesh-wide, so they never take the module's last open ask from a
|
||||
proposal. An answer is the operator's word to the session and authorises nothing; a question never defaults.
|
||||
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 two minutes
|
||||
after it was asked (its refusal not heard here) frees its slot and reads as ended. 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,
|
||||
|
||||
@@ -22,16 +22,16 @@ type Proposing struct {
|
||||
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 ProposalStore
|
||||
Questions QuestionStore
|
||||
questionsMu sync.Mutex
|
||||
}
|
||||
|
||||
// questions is the `questions` state.
|
||||
func (d *Proposing) questions() ProposalStore {
|
||||
func (d *Proposing) questions() QuestionStore {
|
||||
d.questionsMu.Lock()
|
||||
defer d.questionsMu.Unlock()
|
||||
if d.Questions == nil {
|
||||
d.Questions = stateOf{stdio.State(QuestionsState)}
|
||||
d.Questions = keptQuestions{stdio.State(QuestionsState)}
|
||||
}
|
||||
return d.Questions
|
||||
}
|
||||
@@ -154,11 +154,13 @@ func questionTools(p Paths, notLaunched error) []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, and when it expires (5 to 1430 minutes, 60 when " +
|
||||
"absent). Answers at once with the question's key; read the answer with claude_code_answer and keep working " +
|
||||
"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). At most two questions wait at once, mesh-wide. The answer is the " +
|
||||
"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?'"),
|
||||
@@ -171,7 +173,7 @@ func questionTools(p Paths, notLaunched error) []stdio.Tool {
|
||||
}, "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("your session's name, as who@node or who, e.g. fix-589@g14"),
|
||||
"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"},
|
||||
},
|
||||
@@ -191,7 +193,7 @@ func questionTools(p Paths, notLaunched error) []stdio.Tool {
|
||||
"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 at most once a minute.",
|
||||
"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"},
|
||||
|
||||
@@ -27,6 +27,7 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
)
|
||||
|
||||
@@ -37,9 +38,25 @@ const QuestionsState = "questions"
|
||||
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 one is kept for a proposed instruction section (ADR 0274), so a proposal always finds room.
|
||||
// (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
|
||||
|
||||
// RouterHoldsWithin is how long after a question is asked the router has recorded it: it records an ask before it
|
||||
// takes its work. A question it holds no record of after that was refused (and the refusal not heard here) or never
|
||||
// taken: it no longer holds a slot, and reads as ended unless a record of it appears after all.
|
||||
var RouterHoldsWithin = 2 * time.Minute
|
||||
|
||||
// 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 (
|
||||
@@ -71,6 +88,35 @@ type QuestionOption struct {
|
||||
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 {
|
||||
@@ -205,11 +251,8 @@ func QuestionProblem(q Question) string {
|
||||
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)
|
||||
}
|
||||
for _, text := range []string{q.Headline, q.Context, q.Why, q.OnExpiry} {
|
||||
if what := invisible(text); what != "" {
|
||||
add("its words hold %s, which a phone does not show", what)
|
||||
break
|
||||
}
|
||||
if problem := questionWordsProblem(q); problem != "" {
|
||||
add("%s", problem)
|
||||
}
|
||||
if len(problems) == 0 {
|
||||
return ""
|
||||
@@ -217,6 +260,30 @@ func QuestionProblem(q Question) string {
|
||||
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[:])
|
||||
@@ -234,9 +301,9 @@ type QuestionInput struct {
|
||||
Expires time.Duration
|
||||
}
|
||||
|
||||
// AskQuestion keeps a question and asks the operator. It answers at once with the question's key; the answer is read
|
||||
// with ReadAnswer.
|
||||
func AskQuestion(p Paths, in QuestionInput, store ProposalStore, bus AskBus, now time.Time) (map[string]any, error) {
|
||||
// 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 {
|
||||
@@ -246,7 +313,7 @@ func AskQuestion(p Paths, in QuestionInput, store ProposalStore, bus AskBus, now
|
||||
by = who
|
||||
}
|
||||
if by == "" {
|
||||
by = "unnamed"
|
||||
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 {
|
||||
@@ -262,21 +329,25 @@ func AskQuestion(p Paths, in QuestionInput, store ProposalStore, bus AskBus, now
|
||||
if problem := QuestionProblem(q); problem != "" {
|
||||
return refuse(problem)
|
||||
}
|
||||
if open := openQuestions(store, bus, now); len(open) >= MostQuestions {
|
||||
return refuse(fmt.Sprintf("%d questions already wait for the operator's answer (%s), and sessions hold at most %d at "+
|
||||
"once so a proposed instruction section always finds room; read them with claude_code_answer, or wait for "+
|
||||
"one to end", len(open), strings.Join(open, ", "), MostQuestions))
|
||||
}
|
||||
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 {
|
||||
q.Asked = "not asked: " + err.Error()
|
||||
_ = store.Put(q.ID, q)
|
||||
_ = 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)
|
||||
@@ -310,43 +381,99 @@ func labelsOf(q Question) string {
|
||||
return strings.Join(out, ", ")
|
||||
}
|
||||
|
||||
// openQuestions are the keys of the questions still waiting: asked, not ended, not expired.
|
||||
func openQuestions(store ProposalStore, bus AskBus, now time.Time) []string {
|
||||
keys, err := store.Keys()
|
||||
// questionOpen says whether a question still waits: kept, asked, not ended, not expired, and held by the router or
|
||||
// asked less than RouterHoldsWithin ago. 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 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 now.Sub(q.At) < RouterHoldsWithin, 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 _, k := range keys {
|
||||
q, ok := readQuestion(store, k)
|
||||
if !ok || q.Asked != "" || q.Ended != "" || !now.Before(q.Expires) {
|
||||
continue
|
||||
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 entry, err := bus.Record(q.ID); err == nil && entry != nil {
|
||||
var rec routerRecord
|
||||
if json.Unmarshal(entry.Value, &rec) == nil && rec.State != "" && rec.State != "open" {
|
||||
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
|
||||
}
|
||||
}
|
||||
open = append(open, q.ID)
|
||||
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
|
||||
return open, nil
|
||||
}
|
||||
|
||||
func readQuestion(store ProposalStore, key string) (Question, bool) {
|
||||
// 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
|
||||
return Question{}, false, nil
|
||||
}
|
||||
raw, found, err := store.Get(key)
|
||||
raw, _, found, err := store.Get(key)
|
||||
if err != nil || !found {
|
||||
return Question{}, false
|
||||
return Question{}, false, err
|
||||
}
|
||||
var q Question
|
||||
if json.Unmarshal(raw, &q) != nil || q.ID != key {
|
||||
return Question{}, false
|
||||
return Question{}, false, nil
|
||||
}
|
||||
return q, true
|
||||
return q, true, nil
|
||||
}
|
||||
|
||||
// The router's word that a question ended without a choice, as heard on the decided subject.
|
||||
@@ -369,9 +496,10 @@ func forgetQuestionEnd(id string) {
|
||||
}
|
||||
|
||||
// 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 choice is
|
||||
// never taken from this word: only from the router's record, held to the question (ReadAnswer).
|
||||
func HeardQuestionEnd(store ProposalStore, w asks.Warrant) {
|
||||
// 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
|
||||
}
|
||||
@@ -380,9 +508,14 @@ func HeardQuestionEnd(store ProposalStore, w asks.Warrant) {
|
||||
return
|
||||
}
|
||||
if q.Ended == "" {
|
||||
q.Ended = string(w.Outcome)
|
||||
ended := string(w.Outcome)
|
||||
if w.Words != "" {
|
||||
q.Ended += ": " + w.Words
|
||||
ended += ": " + w.Words
|
||||
}
|
||||
if w.Outcome == asks.OutcomeRefused {
|
||||
q = tombstone(q, ended)
|
||||
} else {
|
||||
q.Ended = ended
|
||||
}
|
||||
_ = store.Put(q.ID, q)
|
||||
}
|
||||
@@ -401,7 +534,7 @@ const (
|
||||
)
|
||||
|
||||
// ReadAnswer is where a question stands: waiting, chosen (and what), ended without a choice, or not proven.
|
||||
func ReadAnswer(store ProposalStore, bus AskBus, key string, now time.Time) (map[string]any, error) {
|
||||
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)
|
||||
@@ -426,8 +559,9 @@ func ReadAnswer(store ProposalStore, bus AskBus, key string, now time.Time) (map
|
||||
switch {
|
||||
case entry == nil && q.Ended != "":
|
||||
out["state"], out["words"] = AnswerEnded, q.Ended
|
||||
case entry == nil && now.After(q.Expires.Add(time.Hour)):
|
||||
out["state"], out["words"] = AnswerEnded, "the router keeps no record of it: refused before it was taken, or forgotten"
|
||||
case entry == nil && now.Sub(q.At) >= RouterHoldsWithin:
|
||||
out["state"], out["words"] = AnswerEnded, "the router holds no record of it: refused (its refusal was not heard "+
|
||||
"here) or never taken; nothing waits for the operator's answer"
|
||||
case entry == nil, rec.State == "open", rec.Warrant == nil:
|
||||
out["state"] = AnswerWaiting
|
||||
if q.Ended != "" {
|
||||
@@ -472,7 +606,7 @@ func ReadAnswer(store ProposalStore, bus AskBus, key string, now time.Time) (map
|
||||
}
|
||||
|
||||
// WaitForAnswer reads the answer, and while it is waiting reads it again until wait has passed.
|
||||
func WaitForAnswer(store ProposalStore, bus AskBus, key string, wait time.Duration, now func() time.Time) (map[string]any, error) {
|
||||
func WaitForAnswer(store QuestionStore, bus AskBus, key string, wait time.Duration, now func() time.Time) (map[string]any, error) {
|
||||
if wait > AnswerWaitsAtMost {
|
||||
wait = AnswerWaitsAtMost
|
||||
}
|
||||
|
||||
@@ -6,10 +6,13 @@ package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
)
|
||||
|
||||
@@ -27,7 +30,7 @@ func mergeTonight() QuestionInput {
|
||||
}
|
||||
}
|
||||
|
||||
func askMergeTonight(t *testing.T, p Paths, store *memProposals, r *router, at time.Time) string {
|
||||
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 {
|
||||
@@ -38,7 +41,7 @@ func askMergeTonight(t *testing.T, p Paths, store *memProposals, r *router, at t
|
||||
|
||||
func TestAQuestionIsAskedUnderTheModulesNameAtTheLevelApprove(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
store, r := &memProposals{m: map[string]json.RawMessage{}}, newRouter()
|
||||
store, r := newMemQuestions(), newRouter()
|
||||
out, err := AskQuestion(p, mergeTonight(), store, r, t0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -78,7 +81,7 @@ func TestAQuestionIsAskedUnderTheModulesNameAtTheLevelApprove(t *testing.T) {
|
||||
|
||||
func TestTheChosenAnswerIsReadBackByItsKey(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
store, r := &memProposals{m: map[string]json.RawMessage{}}, newRouter()
|
||||
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 {
|
||||
@@ -100,7 +103,7 @@ func TestTheChosenAnswerIsReadBackByItsKey(t *testing.T) {
|
||||
|
||||
func TestAnAnswerThatDoesNotHoldToTheQuestionIsNotProven(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
store, r := &memProposals{m: map[string]json.RawMessage{}}, newRouter()
|
||||
store, r := newMemQuestions(), newRouter()
|
||||
key := askMergeTonight(t, p, store, r, t0)
|
||||
w := r.chose(key, "o1", t0.Add(time.Minute))
|
||||
|
||||
@@ -115,7 +118,7 @@ func TestAnAnswerThatDoesNotHoldToTheQuestionIsNotProven(t *testing.T) {
|
||||
}
|
||||
|
||||
// A warrant for another ask's digest.
|
||||
store2, r2 := &memProposals{m: map[string]json.RawMessage{}}, newRouter()
|
||||
store2, r2 := newMemQuestions(), newRouter()
|
||||
key2 := askMergeTonight(t, p, store2, r2, t0)
|
||||
w2 := r2.chose(key2, "o1", t0.Add(time.Minute))
|
||||
w2.AskDigest = "sha256:0000"
|
||||
@@ -125,7 +128,7 @@ func TestAnAnswerThatDoesNotHoldToTheQuestionIsNotProven(t *testing.T) {
|
||||
}
|
||||
|
||||
// A choice made after the question expired.
|
||||
store3, r3 := &memProposals{m: map[string]json.RawMessage{}}, newRouter()
|
||||
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 {
|
||||
@@ -150,8 +153,19 @@ func TestAQuestionIsRefusedOutsideItsBounds(t *testing.T) {
|
||||
"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 := &memProposals{m: map[string]json.RawMessage{}}, newRouter()
|
||||
store, r := newMemQuestions(), newRouter()
|
||||
in := mergeTonight()
|
||||
in.Options = append([]QuestionOption(nil), in.Options...)
|
||||
change(&in)
|
||||
@@ -166,7 +180,7 @@ func TestAQuestionIsRefusedOutsideItsBounds(t *testing.T) {
|
||||
|
||||
func TestAtMostTwoQuestionsWaitSoAProposalFindsRoom(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
store, r := &memProposals{m: map[string]json.RawMessage{}}, newRouter()
|
||||
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") {
|
||||
@@ -187,7 +201,7 @@ func TestAtMostTwoQuestionsWaitSoAProposalFindsRoom(t *testing.T) {
|
||||
// subject, with no record.
|
||||
type questionRefusingRouter struct {
|
||||
*router
|
||||
store ProposalStore
|
||||
store QuestionStore
|
||||
id string
|
||||
}
|
||||
|
||||
@@ -203,7 +217,7 @@ func (r *questionRefusingRouter) Publish(subject string, body any, id string) er
|
||||
|
||||
func TestTheRoutersRefusalIsSaidAndKept(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
store := &memProposals{m: map[string]json.RawMessage{}}
|
||||
store := newMemQuestions()
|
||||
refusing := &questionRefusingRouter{router: newRouter(), store: store}
|
||||
within, every := RouterAnswersWithin, RouterAnswersEvery
|
||||
RouterAnswersWithin, RouterAnswersEvery = time.Second, 10*time.Millisecond
|
||||
@@ -216,14 +230,19 @@ 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 open := openQuestions(store, refusing, t0.Add(time.Minute)); len(open) != 0 {
|
||||
t.Errorf("a refused question still counts as waiting: %v", open)
|
||||
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 := &memProposals{m: map[string]json.RawMessage{}}, newRouter()
|
||||
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))
|
||||
@@ -235,7 +254,7 @@ func TestAnExpiredQuestionSaysWhatTheSessionDoes(t *testing.T) {
|
||||
func TestTheQuestionToolsAreServed(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
r := newRouter()
|
||||
questions := &memProposals{m: map[string]json.RawMessage{}}
|
||||
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 })
|
||||
@@ -271,7 +290,7 @@ func TestAReadWaitsNoLongerThanItsBound(t *testing.T) {
|
||||
t.Fatalf("a read may wait %s; ADR 0332 says 60 s", AnswerWaitsAtMost)
|
||||
}
|
||||
p, _ := node(t, "g14")
|
||||
store, r := &memProposals{m: map[string]json.RawMessage{}}, newRouter()
|
||||
store, r := newMemQuestions(), newRouter()
|
||||
key := askMergeTonight(t, p, store, r, t0)
|
||||
most, every := AnswerWaitsAtMost, AnswerLooksEvery
|
||||
AnswerWaitsAtMost, AnswerLooksEvery = 100*time.Millisecond, 10*time.Millisecond
|
||||
@@ -285,3 +304,165 @@ func TestAReadWaitsNoLongerThanItsBound(t *testing.T) {
|
||||
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
|
||||
}
|
||||
|
||||
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()
|
||||
defer s.mu.Unlock()
|
||||
if s.failGet != "" && strings.HasPrefix(key, s.failGet) {
|
||||
return nil, 0, false, errors.New("the bus did not answer")
|
||||
}
|
||||
raw, ok := s.m[key]
|
||||
return raw, s.rev[key], 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, frees its slot once the
|
||||
// router has had time to record it, and reads as ended — not open until it expires.
|
||||
func TestAQuestionTheRouterNeverHeldFreesItsSlot(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) // refused, and the refusal never heard here
|
||||
if _, err := AskQuestion(p, mergeTonight(), store, r, t0.Add(time.Minute)); err == nil {
|
||||
t.Fatal("a slot was freed before the router had time to record the question")
|
||||
}
|
||||
if out, _ := ReadAnswer(store, r, first, t0.Add(RouterHoldsWithin)); out["state"] != AnswerEnded {
|
||||
t.Errorf("read as %v", out)
|
||||
}
|
||||
if _, err := AskQuestion(p, mergeTonight(), store, r, t0.Add(RouterHoldsWithin)); err != nil {
|
||||
t.Errorf("the slot of a question the router never held was not freed: %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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user