mesh-catalog/modules/claude-code/cmd/claude-code/questions_test.go

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)
}