Keep a session's wait list in the claude-code module (issue 1629)
A session can end its turn while an Approve it expects is still open and nothing then watches it. claude_code_wait (add/watched/done/said/list) keeps each session's waits in the module's waits state, keyed by its session id and node, and mirrors them whole and atomically to a file under the agent's home that the plugin's coming hooks (1743, 1744) will read without calling back into this module.
This commit is contained in:
jochen
@@ -0,0 +1,266 @@
|
||||
package main
|
||||
|
||||
// A session's wait list (novox/hq ADR 0489, issue 1629).
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
// memWaits is the `waits` state: values with revisions, create-only and compare-and-set, as the bus keeps it.
|
||||
type memWaits struct {
|
||||
mu sync.Mutex
|
||||
m map[string]json.RawMessage
|
||||
rev map[string]uint64
|
||||
n uint64
|
||||
}
|
||||
|
||||
func newMemWaits() *memWaits {
|
||||
return &memWaits{m: map[string]json.RawMessage{}, rev: map[string]uint64{}}
|
||||
}
|
||||
|
||||
func (s *memWaits) Get(key string) (json.RawMessage, uint64, bool, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
raw, ok := s.m[key]
|
||||
return raw, s.rev[key], ok, nil
|
||||
}
|
||||
func (s *memWaits) 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 *memWaits) 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 *memWaits) 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)
|
||||
}
|
||||
|
||||
// addApproveWait adds the wait fixture the operator's Approve is kept under.
|
||||
func addApproveWait(t *testing.T, p Paths, store WaitStore, session string, at time.Time) string {
|
||||
t.Helper()
|
||||
out, err := AddWait(p, WaitInput{Session: session, By: "fix-1629@" + p.Node, What: "the operator's Approve of pull request 42",
|
||||
Read: WaitRead{Address: "mesh-delivery.recheck", Arguments: map[string]any{"pr": 42}}, Family: "mesh-delivery.decided",
|
||||
Bound: 10}, store, at)
|
||||
if err != nil {
|
||||
t.Fatalf("add: %v", err)
|
||||
}
|
||||
return out["id"].(string)
|
||||
}
|
||||
|
||||
// 1629/AC1: add answers an id; list answers the open waits with their age and whether each is past its bound.
|
||||
func TestAddAnswersAnIDAndListShowsItsAgeAndPastBound(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
store := newMemWaits()
|
||||
id := addApproveWait(t, p, store, "sess-1", t0)
|
||||
if !strings.HasPrefix(id, WaitPrefix) {
|
||||
t.Fatalf("id %q does not start with %q", id, WaitPrefix)
|
||||
}
|
||||
out, err := ListWaits(p, "sess-1", store, t0.Add(time.Minute))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
open := out["open"].([]map[string]any)
|
||||
if len(open) != 1 || open[0]["id"] != id || open[0]["past_bound"] != false || open[0]["age_seconds"] != 60 {
|
||||
t.Fatalf("%+v", open)
|
||||
}
|
||||
out, err = ListWaits(p, "sess-1", store, t0.Add(11*time.Minute))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
open = out["open"].([]map[string]any)
|
||||
if open[0]["past_bound"] != true {
|
||||
t.Fatalf("past its 10-minute bound after 11 minutes: %+v", open[0])
|
||||
}
|
||||
}
|
||||
|
||||
// 1629/AC2: watched, done and said change the wait; said is refused for a wait not past its bound; done removes
|
||||
// it from the open list.
|
||||
func TestWatchedDoneAndSaidChangeTheWait(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
store := newMemWaits()
|
||||
id := addApproveWait(t, p, store, "sess-2", t0)
|
||||
|
||||
if _, err := SaidWait(p, "sess-2", id, store, t0.Add(time.Minute)); err == nil || !strings.Contains(err.Error(), "not past its bound") {
|
||||
t.Fatalf("said before the bound: %v", err)
|
||||
}
|
||||
if _, err := WatchedWait(p, "sess-2", id, "mesh_wait on mesh-delivery.decided", store, t0.Add(time.Minute)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
out, _ := ListWaits(p, "sess-2", store, t0.Add(2*time.Minute))
|
||||
open := out["open"].([]map[string]any)
|
||||
if open[0]["watcher"] != "mesh_wait on mesh-delivery.decided" {
|
||||
t.Fatalf("watcher not kept: %+v", open[0])
|
||||
}
|
||||
if _, err := SaidWait(p, "sess-2", id, store, t0.Add(11*time.Minute)); err != nil {
|
||||
t.Fatalf("said past the bound: %v", err)
|
||||
}
|
||||
out, _ = ListWaits(p, "sess-2", store, t0.Add(12*time.Minute))
|
||||
open = out["open"].([]map[string]any)
|
||||
if open[0]["said"] != true {
|
||||
t.Fatalf("said not kept: %+v", open[0])
|
||||
}
|
||||
if _, err := DoneWait(p, "sess-2", id, "approved", store, t0.Add(13*time.Minute)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
out, _ = ListWaits(p, "sess-2", store, t0.Add(14*time.Minute))
|
||||
if len(out["open"].([]map[string]any)) != 0 {
|
||||
t.Fatalf("done did not remove it from the open list: %v", out)
|
||||
}
|
||||
if _, err := DoneWait(p, "sess-2", id, "approved", store, t0.Add(15*time.Minute)); err == nil {
|
||||
t.Fatal("done on an already-done wait was not refused")
|
||||
}
|
||||
}
|
||||
|
||||
// 1629/AC3: every change rewrites the session's waits file whole, atomically; it holds exactly the open waits
|
||||
// and the write time.
|
||||
func TestEveryChangeRewritesTheWaitsFileWhole(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
store := newMemWaits()
|
||||
id1 := addApproveWait(t, p, store, "sess-3", t0)
|
||||
f, err := readWaitsFile(waitsFilePath(p, "sess-3"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if f.Session != "sess-3" || len(f.Waits) != 1 || f.Waits[0].ID != id1 || !f.At.Equal(t0.UTC()) {
|
||||
t.Fatalf("%+v", f)
|
||||
}
|
||||
// Written whole: no .mesh-tmp left beside it.
|
||||
if _, err := os.Stat(waitsFilePath(p, "sess-3") + ".mesh-tmp"); !os.IsNotExist(err) {
|
||||
t.Fatalf("a temp file was left beside it: %v", err)
|
||||
}
|
||||
id2, err := AddWait(p, WaitInput{Session: "sess-3", By: "fix-1629@" + p.Node, What: "a second wait",
|
||||
Read: WaitRead{Address: "mesh-delivery.recheck"}, Bound: 5}, store, t0.Add(time.Minute))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f, err = readWaitsFile(waitsFilePath(p, "sess-3"))
|
||||
if err != nil || len(f.Waits) != 2 || !f.At.Equal(t0.Add(time.Minute).UTC()) {
|
||||
t.Fatalf("%+v %v", f, err)
|
||||
}
|
||||
if _, err := DoneWait(p, "sess-3", id1, "approved", store, t0.Add(2*time.Minute)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f, err = readWaitsFile(waitsFilePath(p, "sess-3"))
|
||||
if err != nil || len(f.Waits) != 1 || f.Waits[0].ID != id2["id"] {
|
||||
t.Fatalf("the file does not hold exactly the open waits: %+v %v", f, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1629/AC4: the waits are read again after the module "restarts" (the state is kept) and for the same session
|
||||
// after a /clear (the same session id lists them); a different session id lists none of them.
|
||||
func TestWaitsSurviveARestartAndAClearButNotCrossSessions(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
store := newMemWaits()
|
||||
addApproveWait(t, p, store, "sess-4", t0)
|
||||
|
||||
// A restart: a fresh Waiting over the same store reads the same waits.
|
||||
restarted := &Waiting{Store: store, Now: func() time.Time { return t0.Add(time.Minute) }}
|
||||
_ = restarted
|
||||
out, err := ListWaits(p, "sess-4", store, t0.Add(time.Minute))
|
||||
if err != nil || len(out["open"].([]map[string]any)) != 1 {
|
||||
t.Fatalf("not read again after a restart: %v %v", out, err)
|
||||
}
|
||||
|
||||
// A /clear: the same session id still lists it.
|
||||
out, err = ListWaits(p, "sess-4", store, t0.Add(2*time.Minute))
|
||||
if err != nil || len(out["open"].([]map[string]any)) != 1 {
|
||||
t.Fatalf("not read again for the same session: %v %v", out, err)
|
||||
}
|
||||
|
||||
// A different session id: none of them.
|
||||
out, err = ListWaits(p, "sess-other", store, t0.Add(2*time.Minute))
|
||||
if err != nil || len(out["open"].([]map[string]any)) != 0 {
|
||||
t.Fatalf("another session id saw them: %v %v", out, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1629/AC5: add refuses an empty what, a bound under 1 or over 1440 minutes, a read address that is not an
|
||||
// address, and more than 50 open waits per session, naming the argument.
|
||||
func TestAddRefusesOutsideItsBounds(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
base := func() WaitInput {
|
||||
return WaitInput{Session: "sess-5", By: "fix-1629@" + p.Node, What: "something", Read: WaitRead{Address: "mesh-delivery.recheck"}, Bound: 10}
|
||||
}
|
||||
for name, change := range map[string]struct {
|
||||
mutate func(*WaitInput)
|
||||
want string
|
||||
}{
|
||||
"empty what": {func(in *WaitInput) { in.What = "" }, "what is required"},
|
||||
"bound zero": {func(in *WaitInput) { in.Bound = 0 }, "bound_minutes"},
|
||||
"bound under 1": {func(in *WaitInput) { in.Bound = 0 }, "bound_minutes"},
|
||||
"bound over 1440": {func(in *WaitInput) { in.Bound = 1441 }, "bound_minutes"},
|
||||
"empty address": {func(in *WaitInput) { in.Read.Address = "" }, "read.address"},
|
||||
"address no dot": {func(in *WaitInput) { in.Read.Address = "notanaddress" }, "read.address"},
|
||||
"address has spaces": {func(in *WaitInput) { in.Read.Address = "not an address" }, "read.address"},
|
||||
} {
|
||||
in := base()
|
||||
change.mutate(&in)
|
||||
store := newMemWaits()
|
||||
if _, err := AddWait(p, in, store, t0); err == nil || !strings.Contains(err.Error(), change.want) {
|
||||
t.Errorf("%s: %v", name, err)
|
||||
}
|
||||
}
|
||||
|
||||
// More than 50 open waits per session.
|
||||
store := newMemWaits()
|
||||
for i := 0; i < MostWaits; i++ {
|
||||
if _, err := AddWait(p, base(), store, t0); err != nil {
|
||||
t.Fatalf("wait %d: %v", i, err)
|
||||
}
|
||||
}
|
||||
if _, err := AddWait(p, base(), store, t0); err == nil || !strings.Contains(err.Error(), "at most 50 open waits") {
|
||||
t.Fatalf("the 51st wait was not refused: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1629/AC6: the tool is served through the module's own dispatch.
|
||||
func TestTheWaitToolIsServedThroughTheModulesDispatch(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
store := newMemWaits()
|
||||
waiting = &Waiting{Store: store, Now: func() time.Time { return t0 }}
|
||||
t.Cleanup(func() { waiting = nil })
|
||||
s := serve(t, p, &bus{kept: map[string]map[string]any{}}, memConfig{})
|
||||
|
||||
text, isErr := s.call(t, "claude_code_wait", map[string]any{"op": "add", "session": "sess-6", "by": "fix-1629@g14",
|
||||
"what": "the operator's Approve", "read": map[string]any{"address": "mesh-delivery.recheck"}, "bound_minutes": 10})
|
||||
if isErr {
|
||||
t.Fatalf("add: %s", text)
|
||||
}
|
||||
var added map[string]any
|
||||
if err := json.Unmarshal([]byte(text), &added); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
id, _ := added["id"].(string)
|
||||
if id == "" {
|
||||
t.Fatalf("no id answered: %s", text)
|
||||
}
|
||||
|
||||
text, isErr = s.call(t, "claude_code_wait", map[string]any{"op": "list", "session": "sess-6"})
|
||||
if isErr || !strings.Contains(text, id) {
|
||||
t.Fatalf("list: %v %s", isErr, text)
|
||||
}
|
||||
|
||||
text, isErr = s.call(t, "claude_code_wait", map[string]any{"op": "done", "session": "sess-6", "id": id, "outcome": "approved"})
|
||||
if isErr {
|
||||
t.Fatalf("done: %s", text)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user