Let an agent session ask the operator on Telegram and read the answer (issue 589, hq ADR 0332)
mesh/merge-gate head 9446bec2f969 merged into trunk 9c9a76d220f8 — pass: builds claude-code → ace, g14, novox, shanks; no bus step; every machine composes …
mesh/repo-check head 9446bec2f969 merged into trunk 9c9a76d220f8 — fail: its merge-check.sh failed: --- FAIL: TestTheRoutersRefusalIsSaidAndDiscards (0.08s)
mesh/delivery superseded: a newer head of the same pull request

A session that needs a decision could only ask in a terminal the operator may
not watch. claude_code_ask raises the question under this module's name at the
level approve, so only a channel that proves who answered carries it;
claude_code_answer reads the chosen option back by its key from the router's own
record, held to the question kept in the new questions state.
This commit is contained in:
jochen
2026-10-12 22:44:45 +02:00
parent 9c9a76d220
commit 9446bec2f9
5 changed files with 889 additions and 2 deletions
@@ -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 ProposalStore
questionsMu sync.Mutex
}
// questions is the `questions` state.
func (d *Proposing) questions() ProposalStore {
d.questionsMu.Lock()
defer d.questionsMu.Unlock()
if d.Questions == nil {
d.Questions = stateOf{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,81 @@ 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, 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 " +
"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 " +
"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("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 at most once a minute.",
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
}
@@ -0,0 +1,487 @@
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"
"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 one is kept for a proposed instruction section (ADR 0274), so a proposal always finds room.
const MostQuestions = asks.MostOpen - 1
// 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"`
}
// 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)
}
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 len(problems) == 0 {
return ""
}
return strings.Join(problems, "; ")
}
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 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) {
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 == "" {
by = "unnamed"
}
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)
}
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())
}
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 err := bus.Publish(asks.AskSubject(Module), ask, ask.ID); err != nil {
q.Asked = "not asked: " + err.Error()
_ = store.Put(q.ID, q)
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, ", ")
}
// 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()
if err != nil {
return nil
}
var open []string
for _, k := range keys {
q, ok := readQuestion(store, k)
if !ok || q.Asked != "" || q.Ended != "" || !now.Before(q.Expires) {
continue
}
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" {
continue
}
}
open = append(open, q.ID)
}
sort.Strings(open)
return open
}
func readQuestion(store ProposalStore, key string) (Question, bool) {
if !strings.HasPrefix(key, QuestionPrefix) {
return Question{}, false
}
raw, found, err := store.Get(key)
if err != nil || !found {
return Question{}, false
}
var q Question
if json.Unmarshal(raw, &q) != nil || q.ID != key {
return Question{}, false
}
return q, true
}
// 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 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) {
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 == "" {
q.Ended = string(w.Outcome)
if w.Words != "" {
q.Ended += ": " + w.Words
}
_ = 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 ProposalStore, 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.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, 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 ProposalStore, 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)
}
}
@@ -0,0 +1,287 @@
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"
"strings"
"testing"
"time"
"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 *memProposals, 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 := &memProposals{m: map[string]json.RawMessage{}}, 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 := &memProposals{m: map[string]json.RawMessage{}}, 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 := &memProposals{m: map[string]json.RawMessage{}}, 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 := &memProposals{m: map[string]json.RawMessage{}}, 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 := &memProposals{m: map[string]json.RawMessage{}}, 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" },
} {
store, r := &memProposals{m: map[string]json.RawMessage{}}, 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 := &memProposals{m: map[string]json.RawMessage{}}, 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 ProposalStore
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 := &memProposals{m: map[string]json.RawMessage{}}
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 open := openQuestions(store, refusing, t0.Add(time.Minute)); len(open) != 0 {
t.Errorf("a refused question still counts as waiting: %v", open)
}
}
func TestAnExpiredQuestionSaysWhatTheSessionDoes(t *testing.T) {
p, _ := node(t, "g14")
store, r := &memProposals{m: map[string]json.RawMessage{}}, 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 := &memProposals{m: map[string]json.RawMessage{}}
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 := &memProposals{m: map[string]json.RawMessage{}}, 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)
}
}