Files
mesh-catalog/modules/claude-code/cmd/claude-code/waits.go
T
jochen 0f159a1662
mesh/merge-gate head 0f159a166200 merged into trunk 7a7a4034852a — pass: builds claude-code → ace, g14, novox, shanks; no bus step; every machine composes …
mesh/repo-check head 0f159a166200 merged into trunk 7a7a4034852a — pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-…
mesh/delivery delivered
Fix review round 2 of the wait list and its hooks (issues 1629, 1743, 1744, ADR 0490)
The Stop hook treated a missing waits file exactly like one that could
not be read: a notice on every stop of every session with no waits —
the normal state, not an exceptional one. DecideStop now treats
os.ErrNotExist as the ordinary case of no open waits, silently; the
notice is still said for a file that exists but cannot be read or does
not parse.

ADR 0490 §1 keeps waits under a name, but the file a hook reads is kept
per session id: writing it from only the name an op just acted on meant
a second name bound to the same session id had its waits silently drop
out of enforcement the moment the first name was touched, and adopting
one name away emptied the id's file outright even when another name was
still bound to it. Every write now rewrites a session's file from the
union of every name currently bound to it (sessionWaitsUnion), adopt's
rewrite of the superseded id included.

Also: each atomic write gets its own temp file name, so two writes to
the same path can never interleave on one; adopt clears the refusal
counter of both the superseded and the adopting session, so a fresh
three-strikes count starts under the new binding; and list is read-only
again — it no longer claims an unbound name or refuses a session that
is not (yet) the one bound to it, and instead says who currently holds
the name.
2026-10-21 18:43:03 +02:00

777 lines
31 KiB
Go

package main
// A session's wait list (novox/hq ADR 0489, as changed by ADR 0490): what a session still expects an answer
// to — an Approve, a delivery's end, anything it started and must act on when it ends
// (follow-what-you-start) — so the plugin's Stop hook can refuse to end its turn while one is open and
// nothing watches it.
//
// Kept in the module's `waits` state under the session's name and this node (ADR 0490 §1) — `by`, as
// claude_code_ask takes it, never the session id, which Claude Code hands a session a new one of at a
// /clear. Every call also gives `session`, the caller's current id; the module keeps the binding of a name
// to the session that last called for it, with that call's time, so the same name's waits are found again
// across a /clear once the new session has adopted it.
//
// The waits file a hook reads is kept per session id, not per name, and holds the waits of whichever name is
// currently bound to that id (empty for an id nothing is bound to): written whole, atomically, on every
// change, and rewritten for both ids when a binding moves (AdoptName). A second file, the node-wide index,
// lists every name on this node that still holds an open wait, so the SessionStart hook can tell a session
// with a brand new id what it might want to adopt, without calling back into this module.
//
// Every file lives under the agent's home (the manifest's `agent-home` resource), which this process —
// launched as the operator's account (novox/hq ADR 0193) — cannot itself write when the node names an agent
// account of its own (novox/hq ADR 0266): every write below goes through `fsys` (homefs.go), which already
// carries such a write through the sudo helper the guard's binary also serves, exactly as grant.go's
// WriteCredentials does for the agent's own credentials file. Where the node has no agent account of its
// own, fsys is the plain filesystem and the write reaches the operator's own home directly.
import (
"crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"regexp"
"sort"
"strings"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// WaitsState is the module's state the wait lists are kept in.
const WaitsState = "waits"
// WaitsDir is where the mirrored files live under the agent's home.
const WaitsDir = "nox-mesh/waits"
// WaitPrefix starts every wait's id.
const WaitPrefix = "w-"
// MostWaits is how many open waits one name may hold at once (novox/hq ADR 0489, 1629/AC5).
const MostWaits = 50
// The bounds of a wait's bound, in minutes (1629/AC5).
const (
ShortestWaitBound = 1
LongestWaitBound = 1440
)
// mostWaitAttempts bounds the compare-and-set retry loop over the `waits` state.
const mostWaitAttempts = 20
var waitSessionID = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$`)
// requireSessionID trims session and refuses it unless it is a plain id matching waitSessionID — every op
// holds to this (review round 1, item b), so a session id is never used to build a path from anything but a
// safe name (no "/", so it cannot name a file outside the waits directory).
func requireSessionID(session string) (string, error) {
session = strings.TrimSpace(session)
if !waitSessionID.MatchString(session) {
return "", errors.New("session is required: the id the SessionStart hook named")
}
return session, nil
}
// WaitRead is the verb a session reads to see where one wait stands.
type WaitRead struct {
Address string `json:"address"`
Arguments map[string]any `json:"arguments,omitempty"`
}
// Wait is one answer a session wrote down that it expects (novox/hq ADR 0489 §1-§2). By is the name it was
// added under.
type Wait struct {
ID string `json:"id"`
By string `json:"by"`
What string `json:"what"`
Read WaitRead `json:"read"`
Family string `json:"family,omitempty"`
Bound int `json:"bound"` // minutes
At time.Time `json:"at"`
Watcher string `json:"watcher,omitempty"`
Said bool `json:"said,omitempty"`
}
// PastBound says whether now is past this wait's bound.
func (w Wait) PastBound(now time.Time) bool {
return now.Sub(w.At) > time.Duration(w.Bound)*time.Minute
}
// NamedWaits is what the `waits` state keeps for one name on this node (novox/hq ADR 0490 §1): its open
// waits, and the binding to whichever session currently answers to that name — empty until the first call
// names it, by add or by adopt.
type NamedWaits struct {
Name string `json:"name"`
Node string `json:"node"`
CurrentSession string `json:"currentSession,omitempty"`
LastCall time.Time `json:"lastCall,omitempty"`
Waits []Wait `json:"waits"`
}
// WaitsFile is what one session id's mirrored file holds, whole, on every change (1629/AC3): the open waits
// of whichever name is currently bound to this session, and the time they were written.
type WaitsFile struct {
Session string `json:"session"`
At time.Time `json:"at"`
Waits []Wait `json:"waits"`
}
// NodeIndexEntry is one name's open waits, as the node-wide index lists it for the SessionStart hook
// (novox/hq ADR 0490 §2).
type NodeIndexEntry struct {
Name string `json:"name"`
Open int `json:"open"`
CurrentSession string `json:"currentSession,omitempty"`
LastCall time.Time `json:"lastCall"`
}
// NodeIndex is the node-wide index file's whole content (novox/hq ADR 0490 §2): every name on this node that
// still holds an open wait, written by the tool on every change a name's waits or binding takes, so the
// SessionStart hook can say "if this is you, adopt" without calling back into this module.
type NodeIndex struct {
Node string `json:"node"`
At time.Time `json:"at"`
Names []NodeIndexEntry `json:"names"`
}
// WaitStore is the `waits` state: a value with its revision, created only when absent, updated only at the
// revision read; Keys lists every name kept, for the node-wide index.
type WaitStore interface {
Get(key string) (json.RawMessage, uint64, bool, error)
Create(key string, value any) error
Update(key string, value any, revision uint64) error
Keys() ([]string, error)
}
// keptWaits is the `waits` state on the bus.
type keptWaits struct{ s stdio.KeptState }
func (k keptWaits) 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 keptWaits) Create(key string, value any) error {
_, err := k.s.Create(key, value)
return err
}
func (k keptWaits) Update(key string, value any, revision uint64) error {
_, err := k.s.Update(key, value, revision)
return err
}
func (k keptWaits) Keys() ([]string, error) { return k.s.Keys() }
// Waiting is what claude_code_wait reaches: the `waits` state and the clock. Nil outside a launch.
type Waiting struct {
Store WaitStore
Now func() time.Time
}
var waiting *Waiting
// waitsKey is the state's key for one name's waits: its name and this node (novox/hq ADR 0490 §1), so two
// nodes never collide over the same name.
func waitsKey(p Paths, name string) string { return p.Node + "/" + name }
// WaitsDirPath is where this node's sessions' waits are mirrored, under the agent's home: the directory the
// plugin's SessionStart and Stop hooks (1744, 1743) are given on their command line.
func WaitsDirPath(p Paths) string {
return filepath.Join(p.Home, ".claude", filepath.FromSlash(WaitsDir))
}
// waitsFilePath is where one session id's mirrored waits are.
func waitsFilePath(p Paths, session string) string {
return filepath.Join(WaitsDirPath(p), session+".json")
}
// nodeIndexPath is the node-wide index's file, beside the per-session ones; its name can never collide with
// a session id, since waitSessionID never allows a leading "_".
func nodeIndexPath(p Paths) string { return filepath.Join(WaitsDirPath(p), "_index.json") }
func newWaitID() string {
var b [8]byte
_, _ = rand.Read(b[:])
return WaitPrefix + hex.EncodeToString(b[:])
}
// addressShaped says whether s could be an address `mesh_call` takes: not blank, no whitespace, and a dot
// between two names (`<seat>.<verb>`, `<node>/<seat>.<verb>`, `<node>/<module>.<tool>`).
func addressShaped(s string) bool {
s = strings.TrimSpace(s)
if s == "" || strings.ContainsAny(s, " \t\n\r") {
return false
}
dot := strings.LastIndex(s, ".")
return dot > 0 && dot < len(s)-1
}
func readNamedWaits(store WaitStore, key string) (NamedWaits, uint64, bool, error) {
raw, rev, found, err := store.Get(key)
if err != nil || !found {
return NamedWaits{}, 0, found, err
}
var nw NamedWaits
_ = json.Unmarshal(raw, &nw)
return nw, rev, true, nil
}
// updateNamedWaits reads one name's waits, lets mutate change them, and writes the result back at the
// revision read — created when there was none yet — retried when another write won the race.
func updateNamedWaits(store WaitStore, key string, mutate func(NamedWaits) (NamedWaits, error)) (NamedWaits, error) {
for i := 0; i < mostWaitAttempts; i++ {
nw, rev, found, err := readNamedWaits(store, key)
if err != nil {
return NamedWaits{}, fmt.Errorf("the wait list could not be read: %w", err)
}
next, err := mutate(nw)
if err != nil {
return NamedWaits{}, err
}
if !found {
err = store.Create(key, next)
} else {
err = store.Update(key, next, rev)
}
if err == nil {
return next, nil
}
if errors.Is(err, stdio.ErrChanged) {
continue
}
return NamedWaits{}, fmt.Errorf("the wait list could not be written: %w", err)
}
return NamedWaits{}, errors.New("the wait list kept changing elsewhere; try again")
}
// tempSuffix is a short random name for a temp file, so two writes racing for the same path never share one
// (review round 2, item a): each gets its own, and only the rename is a single atomic step.
func tempSuffix() string {
var b [8]byte
_, _ = rand.Read(b[:])
return hex.EncodeToString(b[:])
}
// writeSessionFile rewrites one session id's mirrored file whole, atomically (its own temp file, then
// rename): it holds exactly the open waits given and the time of this write (1629/AC3). waits is always the
// union of every name currently bound to sessionID (sessionWaitsUnion), never only the name an op acted on:
// a second name bound to the same id must stay enforced (ADR 0490 §1, review round 2 item 2).
func writeSessionFile(p Paths, sessionID string, waits []Wait, now time.Time) error {
path := waitsFilePath(p, sessionID)
if err := fsys.MkdirAll(filepath.Dir(path), 0o700); err != nil {
return err
}
if waits == nil {
waits = []Wait{}
}
raw, err := indented(WaitsFile{Session: sessionID, At: now.UTC(), Waits: waits})
if err != nil {
return err
}
tmp := path + ".mesh-tmp-" + tempSuffix()
if err := fsys.WriteFile(tmp, raw, 0o600); err != nil {
return err
}
return fsys.Rename(tmp, path)
}
// sessionWaitsUnion is every open wait of every name on this node currently bound to session (ADR 0490 §1):
// what that session id's mirrored file must hold, whole — never only the one name an op just acted on, or a
// second name bound to the same id would stop being enforced the moment any op touched the first.
func sessionWaitsUnion(store WaitStore, node, session string) ([]Wait, error) {
keys, err := store.Keys()
if err != nil {
return nil, err
}
prefix := node + "/"
var out []Wait
names := make([]string, 0, len(keys))
for _, k := range keys {
if strings.HasPrefix(k, prefix) {
names = append(names, k)
}
}
sort.Strings(names) // deterministic order: tests, and a session reading its own file, see the same thing
for _, k := range names {
nw, _, found, err := readNamedWaits(store, k)
if err != nil || !found || nw.CurrentSession != session {
continue
}
out = append(out, nw.Waits...)
}
return out, nil
}
// writeSessionWaits rewrites session's mirrored file with sessionWaitsUnion: every call that changes a
// name's waits or its binding goes through this, never writeSessionFile directly with one name's waits
// alone.
func writeSessionWaits(p Paths, store WaitStore, session string, now time.Time) error {
union, err := sessionWaitsUnion(store, p.Node, session)
if err != nil {
return fmt.Errorf("the session's other names could not be read: %w", err)
}
return writeSessionFile(p, session, union, now)
}
// writeNodeIndex rewrites the node-wide index whole, atomically, from every name's current state on this
// node (novox/hq ADR 0490 §2): a full scan of the `waits` state, bounded by how many names are ever used on
// one node at once. Best-effort for its caller: a name's own file is already correct by the time this runs.
func writeNodeIndex(p Paths, store WaitStore, now time.Time) error {
keys, err := store.Keys()
if err != nil {
return fmt.Errorf("the node-wide wait index could not be read: %w", err)
}
prefix := p.Node + "/"
idx := NodeIndex{Node: p.Node, At: now.UTC(), Names: []NodeIndexEntry{}}
for _, k := range keys {
name, ok := strings.CutPrefix(k, prefix)
if !ok {
continue
}
nw, _, found, err := readNamedWaits(store, k)
if err != nil || !found || len(nw.Waits) == 0 {
continue
}
idx.Names = append(idx.Names, NodeIndexEntry{Name: name, Open: len(nw.Waits),
CurrentSession: nw.CurrentSession, LastCall: nw.LastCall.UTC()})
}
sort.Slice(idx.Names, func(i, j int) bool { return idx.Names[i].Name < idx.Names[j].Name })
path := nodeIndexPath(p)
if err := fsys.MkdirAll(filepath.Dir(path), 0o700); err != nil {
return err
}
raw, err := indented(idx)
if err != nil {
return err
}
tmp := path + ".mesh-tmp-" + tempSuffix()
if err := fsys.WriteFile(tmp, raw, 0o600); err != nil {
return err
}
return fsys.Rename(tmp, path)
}
// callerName reads `by` as claude_code_ask does: who@node, where node is this one — the name a wait list is
// kept under (novox/hq ADR 0490 §1), never the session id.
func callerName(p Paths, by string) (string, error) {
by = strings.TrimSpace(by)
if who, node, found := strings.Cut(by, "@"); found {
if node != p.Node {
return "", fmt.Errorf("by names the node %q, and this is %s: call on the node the session runs on", node, p.Node)
}
return who, nil
}
if by == "" {
return "", errors.New("by is required: your session's name, as who@node or who")
}
return by, nil
}
// bindCaller checks the binding for name against session, refusing when another, more recently bound
// session holds it (novox/hq ADR 0490 §3) — naming that session and when it last called, and how to take the
// name back with adopt — and otherwise (re)binds it to session at now, so a name nothing has bound yet is
// claimed by the first session that calls for it, add included.
func bindCaller(nw *NamedWaits, name, node, by, session string, now time.Time) error {
if nw.CurrentSession != "" && nw.CurrentSession != session {
return fmt.Errorf("the name %q now answers to session %s (it last called at %s); adopt it again if this "+
"is you: claude_code_wait {\"op\":\"adopt\",\"by\":%q,\"session\":%q}", name, nw.CurrentSession,
nw.LastCall.UTC().Format(time.RFC3339), by, session)
}
nw.Name, nw.Node, nw.CurrentSession, nw.LastCall = name, node, session, now.UTC()
return nil
}
// findWait is the index of id in waits, or -1.
func findWait(waits []Wait, id string) int {
for i, w := range waits {
if w.ID == id {
return i
}
}
return -1
}
// WaitInput is what claude_code_wait's add takes.
type WaitInput struct {
Session string
By string
What string
Read WaitRead
Family string
Bound int
}
// AddWait keeps one more wait for a name, refusing outside ADR 0489's bounds (1629/AC5) or when another
// session has since taken the name over (ADR 0490 §3), and rewrites the caller's mirrored file (1629/AC3).
func AddWait(p Paths, in WaitInput, store WaitStore, now time.Time) (map[string]any, error) {
session, err := requireSessionID(in.Session)
if err != nil {
return nil, err
}
name, err := callerName(p, in.By)
if err != nil {
return nil, err
}
what := strings.TrimSpace(in.What)
if what == "" {
return nil, errors.New("what is required: say what you are waiting for")
}
if in.Bound < ShortestWaitBound || in.Bound > LongestWaitBound {
return nil, fmt.Errorf("bound_minutes is %d; it is %d to %d", in.Bound, ShortestWaitBound, LongestWaitBound)
}
address := strings.TrimSpace(in.Read.Address)
if !addressShaped(address) {
return nil, fmt.Errorf("read.address %q is not an address: the verb that reads where this wait stands, as mesh_call takes it", in.Read.Address)
}
w := Wait{ID: newWaitID(), By: name, What: what, Read: WaitRead{Address: address, Arguments: in.Read.Arguments},
Family: strings.TrimSpace(in.Family), Bound: in.Bound, At: now.UTC()}
key := waitsKey(p, name)
nw, err := updateNamedWaits(store, key, func(nw NamedWaits) (NamedWaits, error) {
if err := bindCaller(&nw, name, p.Node, in.By, session, now); err != nil {
return nw, err
}
if len(nw.Waits) >= MostWaits {
return nw, fmt.Errorf("%d waits already wait for %s; at most %d open waits per name", len(nw.Waits), name, MostWaits)
}
nw.Waits = append(append([]Wait(nil), nw.Waits...), w)
return nw, nil
})
if err != nil {
return nil, err
}
if err := writeSessionWaits(p, store, session, now); err != nil {
return nil, fmt.Errorf("the wait was kept, but its file could not be written: %w", err)
}
if err := writeNodeIndex(p, store, now); err != nil {
say("writing the node-wide wait index after %s added for %s: %v", w.ID, name, err)
}
say("wait %s added for %s (session %s): %s", w.ID, name, session, what)
return map[string]any{"id": w.ID, "name": name, "session": session, "open": len(nw.Waits)}, nil
}
// WatchedWait marks a wait as followed: by mesh_wait on its family, or a watcher started for it (1629/AC2).
func WatchedWait(p Paths, session, by, id, watcher string, store WaitStore, now time.Time) (map[string]any, error) {
session, err := requireSessionID(session)
if err != nil {
return nil, err
}
name, err := callerName(p, by)
if err != nil {
return nil, err
}
id, watcher = strings.TrimSpace(id), strings.TrimSpace(watcher)
if id == "" {
return nil, errors.New("id is required: the one add answered")
}
if watcher == "" {
return nil, errors.New("watcher is required: what follows this wait (mesh_wait on its family, or a watcher agent)")
}
key := waitsKey(p, name)
_, err = updateNamedWaits(store, key, func(nw NamedWaits) (NamedWaits, error) {
if err := bindCaller(&nw, name, p.Node, by, session, now); err != nil {
return nw, err
}
i := findWait(nw.Waits, id)
if i < 0 {
return nw, fmt.Errorf("no open wait %s for %s", id, name)
}
nw.Waits[i].Watcher = watcher
return nw, nil
})
if err != nil {
return nil, err
}
if err := writeSessionWaits(p, store, session, now); err != nil {
return nil, fmt.Errorf("the wait was watched, but its file could not be written: %w", err)
}
_ = writeNodeIndex(p, store, now)
return map[string]any{"id": id, "watcher": watcher}, nil
}
// DoneWait removes a wait from the open list: it ended, with outcome (1629/AC2).
func DoneWait(p Paths, session, by, id, outcome string, store WaitStore, now time.Time) (map[string]any, error) {
session, err := requireSessionID(session)
if err != nil {
return nil, err
}
name, err := callerName(p, by)
if err != nil {
return nil, err
}
id, outcome = strings.TrimSpace(id), strings.TrimSpace(outcome)
if id == "" {
return nil, errors.New("id is required: the one add answered")
}
key := waitsKey(p, name)
_, err = updateNamedWaits(store, key, func(nw NamedWaits) (NamedWaits, error) {
if err := bindCaller(&nw, name, p.Node, by, session, now); err != nil {
return nw, err
}
i := findWait(nw.Waits, id)
if i < 0 {
return nw, fmt.Errorf("no open wait %s for %s", id, name)
}
nw.Waits = append(append([]Wait(nil), nw.Waits[:i]...), nw.Waits[i+1:]...)
return nw, nil
})
if err != nil {
return nil, err
}
if err := writeSessionWaits(p, store, session, now); err != nil {
return nil, fmt.Errorf("the wait was done, but its file could not be written: %w", err)
}
_ = writeNodeIndex(p, store, now)
say("wait %s for %s (session %s) done: %s", id, name, session, outcome)
return map[string]any{"id": id, "done": true, "outcome": outcome}, nil
}
// SaidWait marks a wait past its bound as told to the operator: refused while it is still within its bound
// (1629/AC2).
func SaidWait(p Paths, session, by, id string, store WaitStore, now time.Time) (map[string]any, error) {
session, err := requireSessionID(session)
if err != nil {
return nil, err
}
name, err := callerName(p, by)
if err != nil {
return nil, err
}
id = strings.TrimSpace(id)
if id == "" {
return nil, errors.New("id is required: the one add answered")
}
key := waitsKey(p, name)
_, err = updateNamedWaits(store, key, func(nw NamedWaits) (NamedWaits, error) {
if err := bindCaller(&nw, name, p.Node, by, session, now); err != nil {
return nw, err
}
i := findWait(nw.Waits, id)
if i < 0 {
return nw, fmt.Errorf("no open wait %s for %s", id, name)
}
if !nw.Waits[i].PastBound(now) {
return nw, fmt.Errorf("wait %s is not past its bound yet: said is for telling the operator about one that is", id)
}
nw.Waits[i].Said = true
return nw, nil
})
if err != nil {
return nil, err
}
if err := writeSessionWaits(p, store, session, now); err != nil {
return nil, fmt.Errorf("the wait was said, but its file could not be written: %w", err)
}
_ = writeNodeIndex(p, store, now)
return map[string]any{"id": id, "said": true}, nil
}
// ListWaits answers a name's open waits, with their age and whether they are past their bound (1629/AC1),
// and who currently holds the name. Read-only (review round 2, item c): unlike every other op it never
// claims an unbound name, and never refuses a session that is not (or not yet) the one bound to it — a
// session may want to look before adopting.
func ListWaits(p Paths, session, by string, store WaitStore, now time.Time) (map[string]any, error) {
session, err := requireSessionID(session)
if err != nil {
return nil, err
}
name, err := callerName(p, by)
if err != nil {
return nil, err
}
nw, _, _, err := readNamedWaits(store, waitsKey(p, name))
if err != nil {
return nil, fmt.Errorf("the wait list could not be read: %w", err)
}
out := []map[string]any{}
for _, w := range nw.Waits {
entry := map[string]any{"id": w.ID, "what": w.What, "by": w.By, "address": w.Read.Address,
"bound_minutes": w.Bound, "at": w.At.Format(time.RFC3339), "age_seconds": int(now.Sub(w.At).Seconds()),
"past_bound": w.PastBound(now), "watcher": w.Watcher, "said": w.Said}
if len(w.Read.Arguments) > 0 {
entry["arguments"] = w.Read.Arguments
}
if w.Family != "" {
entry["family"] = w.Family
}
out = append(out, entry)
}
answer := map[string]any{"session": session, "name": name, "open": out}
if nw.CurrentSession != "" {
answer["held-by"] = map[string]any{"session": nw.CurrentSession, "last-called": nw.LastCall.UTC().Format(time.RFC3339)}
}
return answer, nil
}
// AdoptName binds name to session, always taking it over from whoever held it (novox/hq ADR 0490 §3): the
// mirrored files of both the session it is taken from and the one adopting it are rewritten at once, so a
// session that no longer owns a name never finds its old waits in its own file, and the new one does right
// away — no waiting on a bound, unlike the Stop and SessionStart hooks' own.
func AdoptName(p Paths, by, session string, store WaitStore, now time.Time) (map[string]any, error) {
session, err := requireSessionID(session)
if err != nil {
return nil, err
}
name, err := callerName(p, by)
if err != nil {
return nil, err
}
key := waitsKey(p, name)
var oldSession string
var oldLastCall time.Time
nw, err := updateNamedWaits(store, key, func(nw NamedWaits) (NamedWaits, error) {
oldSession, oldLastCall = nw.CurrentSession, nw.LastCall
nw.Name, nw.Node, nw.CurrentSession, nw.LastCall = name, p.Node, session, now.UTC()
return nw, nil
})
if err != nil {
return nil, err
}
tookOver := oldSession != "" && oldSession != session
if tookOver {
// oldSession's file is rewritten from the union of whatever names remain bound to it (ADR 0490 §1,
// review round 2 item 2): name no longer is one of them, but another might still be — never simply
// emptied.
if err := writeSessionWaits(p, store, oldSession, now); err != nil {
return nil, fmt.Errorf("%s was adopted, but the session it was taken from could not be told: %w", name, err)
}
}
if err := writeSessionWaits(p, store, session, now); err != nil {
return nil, fmt.Errorf("the name was adopted, but its file could not be written: %w", err)
}
if err := writeNodeIndex(p, store, now); err != nil {
say("writing the node-wide wait index after %s was adopted: %v", name, err)
}
// Review round 2, item b: a fresh three-strikes count starts under the new binding, on both ids.
waitsDir := WaitsDirPath(p)
_ = writeRefusalCounters(waitsDir, session, map[string]int{})
if tookOver {
_ = writeRefusalCounters(waitsDir, oldSession, map[string]int{})
}
answer := map[string]any{"name": name, "session": session, "open": len(nw.Waits)}
if tookOver {
answer["took-over-from"] = map[string]any{"session": oldSession, "last-called": oldLastCall.UTC().Format(time.RFC3339)}
say("%s adopted by session %s, taken over from %s (last called %s)", name, session, oldSession,
oldLastCall.UTC().Format(time.RFC3339))
} else {
say("%s adopted by session %s", name, session)
}
return answer, nil
}
// waitsFileIn is where one session's mirrored waits are, given a hook's own waits-dir argument: the hooks
// have no launch environment, only their command line, so they never recompute this from Paths.
func waitsFileIn(waitsDir, session string) string { return filepath.Join(waitsDir, session+".json") }
// nodeIndexPathIn is the node-wide index file, given a hook's own waits-dir argument.
func nodeIndexPathIn(waitsDir string) string { return filepath.Join(waitsDir, "_index.json") }
// readWaitsFile reads one session's mirrored file, as the hooks do: a plain local read, as the session's own
// account (never through fsys or the module's state), and never calling back into this module.
func readWaitsFile(path string) (WaitsFile, error) {
raw, err := os.ReadFile(path)
if err != nil {
return WaitsFile{}, err
}
var f WaitsFile
if err := json.Unmarshal(raw, &f); err != nil {
return WaitsFile{}, err
}
return f, nil
}
// readNodeIndex reads the node-wide index, as the SessionStart hook does: a plain local read, never through
// fsys or the module's state, and never calling back into this module.
func readNodeIndex(path string) (NodeIndex, error) {
raw, err := os.ReadFile(path)
if err != nil {
return NodeIndex{}, err
}
var idx NodeIndex
if err := json.Unmarshal(raw, &idx); err != nil {
return NodeIndex{}, err
}
return idx, nil
}
// waitTools is claude_code_wait: add, watched, done, said, list and adopt, over a name's wait list (novox/hq
// ADR 0489, ADR 0490).
func waitTools(p Paths) []stdio.Tool {
notLaunched := errors.New("this module was not launched with the bus; no wait can be kept")
return []stdio.Tool{
{Name: "claude_code_wait",
Description: "Your wait list, kept under your session's name (novox/hq ADR 0489, ADR 0490): what you " +
"still expect an answer to — an Approve, a delivery's end, anything you started and must act on when " +
"it ends. The plugin's Stop hook refuses to end your turn while a wait here is open and nothing " +
"follows it. op is add, watched, done, said, list or adopt; session (your current session id, named " +
"by the SessionStart hook) and by (your session's name, as who@node or who) are always required. A " +
"name is bound to whichever session last called for it; if your own id changed (a /clear) you must " +
"adopt your name again before it answers to you — the SessionStart hook lists every name on this " +
"node with an open wait and how. adopt takes the name over at once, however recently another " +
"session called for it; that session is told so, naming yours, the next time it calls. add takes " +
"what (what you are waiting for), read (the verb that answers where it stands: {address, " +
"arguments}), family (optional, the event mesh_wait can wait on) and bound_minutes (1 to 1440), and " +
"answers an id; at most 50 waits are open per name. watched takes id and watcher (what follows it: " +
"mesh_wait on its family, or a watcher agent you started). done takes id and outcome, and removes it " +
"from the open list. said takes id and marks a wait past its bound as told to the operator — " +
"refused while it is still within its bound. list answers the open waits, with their age and " +
"whether each is past its bound.",
Input: map[string]any{
"op": str("add, watched, done, said, list or adopt"),
"session": str("required: your session's current id"),
"by": str("required: your session's name, as who@node or who"),
"what": str("add: what you are waiting for, in plain words"),
"read": map[string]any{"type": "object", "description": "add: the verb that reads where this wait stands",
"properties": map[string]any{
"address": str("the address, as mesh_call takes it"),
"arguments": map[string]any{"type": "object", "description": "its arguments"},
}},
"family": str("add: the event family mesh_wait can wait on, when there is one"),
"bound_minutes": map[string]any{"type": "integer", "description": "add: minutes until this wait's bound, 1 to 1440"},
"id": str("watched, done or said: the id add answered"),
"watcher": str("watched: what follows this wait"),
"outcome": str("done: what the wait ended with"),
},
Run: func(a map[string]any) (any, error) {
if waiting == nil {
return nil, notLaunched
}
session, by := strArg(a, "session"), strArg(a, "by")
now := time.Now()
if waiting.Now != nil {
now = waiting.Now()
}
switch op := strArg(a, "op"); op {
case "add":
read := WaitRead{}
if m, ok := a["read"].(map[string]any); ok {
read.Address = strArg(m, "address")
if args, ok := m["arguments"].(map[string]any); ok {
read.Arguments = args
}
}
bound := 0
if n, ok := a["bound_minutes"].(float64); ok {
bound = int(n)
}
return AddWait(p, WaitInput{Session: session, By: by, What: strArg(a, "what"), Read: read,
Family: strArg(a, "family"), Bound: bound}, waiting.Store, now)
case "watched":
return WatchedWait(p, session, by, strArg(a, "id"), strArg(a, "watcher"), waiting.Store, now)
case "done":
return DoneWait(p, session, by, strArg(a, "id"), strArg(a, "outcome"), waiting.Store, now)
case "said":
return SaidWait(p, session, by, strArg(a, "id"), waiting.Store, now)
case "list":
return ListWaits(p, session, by, waiting.Store, now)
case "adopt":
return AdoptName(p, by, session, waiting.Store, now)
default:
return nil, fmt.Errorf("op is add, watched, done, said, list or adopt, not %q", op)
}
}},
}
}