novox/mesh-catalog
18
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Key a wait list by name, not session id, so it survives a /clear (issues 1629, 1744, ADR 0490)

Browse Source 
Claude Code gives a session a new session_id at /clear, with nothing
naming the earlier one, so keying claude_code_wait's state by session id
lost a session's waits the moment it cleared. ADR 0490 changes ADR 0489
§1: waits are now kept under the caller's name and node (by, who@node);
every op also takes session, the caller's current id, and the module
keeps a binding of the name to the session that last called for it. An
unbound name is claimed by the first call for it; a session whose own
id no longer matches the binding is refused on every op, naming the
session that holds it, until it calls the new op adopt, which always
takes the name over at once (no bound protects a recent call) and tells
the earlier session so on its own next call.

The mirrored file a hook reads stays keyed by session id (a hook only
ever has that), holding whichever name's waits are currently bound to
it; adopting rewrites both the old id's file (emptied) and the new id's
(filled) in the same call. A second mirrored file, the node-wide index,
lists every name on the node with an open wait; the SessionStart hook
now also reads it, on every start, and prints the exact adopt call for
each name, so a session with a brand new id can find and reclaim its
own waits without calling back into the module.
This commit is contained in:
jochen
2026-10-21 18:32:50 +02:00
parent 02a8837348
commit f0beea153e
4 changed files with 602 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/claude-code/cmd/claude-code/waits.go
+333 -131
View File
@@ -1,18 +1,28 @@
package main
// A session's wait list (novox/hq ADR 0489): 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.
// 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 id and this node (so the same session reads its
// waits again after a /clear, and after the module restarts), and mirrored whole, atomically, to a file the
// hooks read without calling back into this module: <agent home>/.claude/nox-mesh/waits/<session>.json. That
// path is 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.
// 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"
@@ -23,6 +33,7 @@ import (
"os"
"path/filepath"
"regexp"
"sort"
"strings"
"time"
@@ -38,7 +49,7 @@ const WaitsDir = "nox-mesh/waits"
// WaitPrefix starts every wait's id.
const WaitPrefix = "w-"
// MostWaits is how many open waits one session may hold at once (novox/hq ADR 0489, 1629/AC5).
// 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).
@@ -53,8 +64,8 @@ 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, not only add (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).
// 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) {
@@ -69,7 +80,8 @@ type WaitRead struct {
Arguments map[string]any `json:"arguments,omitempty"`
}
// Wait is one answer a session wrote down that it expects (novox/hq ADR 0489 §1-§2).
// 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"`
@@ -87,27 +99,50 @@ func (w Wait) PastBound(now time.Time) bool {
return now.Sub(w.At) > time.Duration(w.Bound)*time.Minute
}
// SessionWaits is what the `waits` state keeps for one session: its open waits.
type SessionWaits struct {
Session string `json:"session"`
Node string `json:"node"`
Waits []Wait `json:"waits"`
// 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 the mirrored file holds, whole, on every change (1629/AC3): the session's open waits and
// the time they were written.
// 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.
// 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.
@@ -128,6 +163,7 @@ 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 {
@@ -137,9 +173,9 @@ type Waiting struct {
var waiting *Waiting
// waitsKey is the state's key for one session's waits: its id and this node (novox/hq ADR 0489 §1), so two
// nodes never collide over the same session id.
func waitsKey(p Paths, session string) string { return p.Node + "/" + session }
// 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.
@@ -147,11 +183,15 @@ func WaitsDirPath(p Paths) string {
return filepath.Join(p.Home, ".claude", filepath.FromSlash(WaitsDir))
}
// waitsFilePath is where one session's waits are mirrored.
// 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[:])
@@ -169,27 +209,27 @@ func addressShaped(s string) bool {
return dot > 0 && dot < len(s)-1
}
func readSessionWaits(store WaitStore, key string) (SessionWaits, uint64, bool, error) {
func readNamedWaits(store WaitStore, key string) (NamedWaits, uint64, bool, error) {
raw, rev, found, err := store.Get(key)
if err != nil || !found {
return SessionWaits{}, 0, found, err
return NamedWaits{}, 0, found, err
}
var sw SessionWaits
_ = json.Unmarshal(raw, &sw)
return sw, rev, true, nil
var nw NamedWaits
_ = json.Unmarshal(raw, &nw)
return nw, rev, true, nil
}
// updateWaits reads one session's waits, lets mutate change them, and writes the result back at the
// 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 updateWaits(store WaitStore, key string, mutate func(SessionWaits) (SessionWaits, error)) (SessionWaits, error) {
func updateNamedWaits(store WaitStore, key string, mutate func(NamedWaits) (NamedWaits, error)) (NamedWaits, error) {
for i := 0; i < mostWaitAttempts; i++ {
sw, rev, found, err := readSessionWaits(store, key)
nw, rev, found, err := readNamedWaits(store, key)
if err != nil {
return SessionWaits{}, fmt.Errorf("the wait list could not be read: %w", err)
return NamedWaits{}, fmt.Errorf("the wait list could not be read: %w", err)
}
next, err := mutate(sw)
next, err := mutate(nw)
if err != nil {
return SessionWaits{}, err
return NamedWaits{}, err
}
if !found {
err = store.Create(key, next)
@@ -202,23 +242,22 @@ func updateWaits(store WaitStore, key string, mutate func(SessionWaits) (Session
if errors.Is(err, stdio.ErrChanged) {
continue
}
return SessionWaits{}, fmt.Errorf("the wait list could not be written: %w", err)
return NamedWaits{}, fmt.Errorf("the wait list could not be written: %w", err)
}
return SessionWaits{}, errors.New("the wait list kept changing elsewhere; try again")
return NamedWaits{}, errors.New("the wait list kept changing elsewhere; try again")
}
// writeWaitsFile rewrites the session's mirrored file whole, atomically (temp file, then rename): it holds
// exactly the open waits and the time of this write (1629/AC3).
func writeWaitsFile(p Paths, session string, sw SessionWaits, now time.Time) error {
path := waitsFilePath(p, session)
// writeSessionFile rewrites one session id's mirrored file whole, atomically (temp file, then rename): it
// holds exactly the open waits given and the time of this write (1629/AC3).
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
}
waits := sw.Waits
if waits == nil {
waits = []Wait{}
}
raw, err := indented(WaitsFile{Session: session, At: now.UTC(), Waits: waits})
raw, err := indented(WaitsFile{Session: sessionID, At: now.UTC(), Waits: waits})
if err != nil {
return err
}
@@ -229,12 +268,51 @@ func writeWaitsFile(p Paths, session string, sw SessionWaits, now time.Time) err
return fsys.Rename(tmp, path)
}
// sessionAndNode reads `by` as claude_code_ask does: who@node, where node is this one.
func sessionAndNode(p Paths, by string) (string, error) {
// 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"
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: add the wait on the node the session runs on", 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
}
@@ -244,6 +322,30 @@ func sessionAndNode(p Paths, by string) (string, error) {
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
@@ -254,14 +356,14 @@ type WaitInput struct {
Bound int
}
// AddWait keeps one more wait for a session, refusing outside ADR 0489's bounds (1629/AC5), and rewrites the
// session's mirrored file (1629/AC3).
// 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
}
by, err := sessionAndNode(p, in.By)
name, err := callerName(p, in.By)
if err != nil {
return nil, err
}
@@ -276,43 +378,42 @@ func AddWait(p Paths, in WaitInput, store WaitStore, now time.Time) (map[string]
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: by, What: what, Read: WaitRead{Address: address, Arguments: in.Read.Arguments},
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, session)
sw, err := updateWaits(store, key, func(sw SessionWaits) (SessionWaits, error) {
if len(sw.Waits) >= MostWaits {
return sw, fmt.Errorf("%d waits already wait for session %s; at most %d open waits per session", len(sw.Waits), session, MostWaits)
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
}
sw.Session, sw.Node = session, p.Node
sw.Waits = append(append([]Wait(nil), sw.Waits...), w)
return sw, nil
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 := writeWaitsFile(p, session, sw, now); err != nil {
if err := writeSessionFile(p, session, nw.Waits, now); err != nil {
return nil, fmt.Errorf("the wait was kept, but its file could not be written: %w", err)
}
say("wait %s added for session %s: %s", w.ID, session, what)
return map[string]any{"id": w.ID, "session": session, "open": len(sw.Waits)}, nil
}
// findWait is the index of id in sw.Waits, or -1.
func findWait(sw SessionWaits, id string) int {
for i, w := range sw.Waits {
if w.ID == id {
return i
}
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)
}
return -1
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, id, watcher string, store WaitStore, now time.Time) (map[string]any, error) {
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")
@@ -320,97 +421,127 @@ func WatchedWait(p Paths, session, id, watcher string, store WaitStore, now time
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, session)
sw, err := updateWaits(store, key, func(sw SessionWaits) (SessionWaits, error) {
i := findWait(sw, id)
if i < 0 {
return sw, fmt.Errorf("no open wait %s for session %s", id, session)
key := waitsKey(p, name)
nw, err := updateNamedWaits(store, key, func(nw NamedWaits) (NamedWaits, error) {
if err := bindCaller(&nw, name, p.Node, by, session, now); err != nil {
return nw, err
}
sw.Waits[i].Watcher = watcher
return sw, nil
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 := writeWaitsFile(p, session, sw, now); err != nil {
if err := writeSessionFile(p, session, nw.Waits, 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, id, outcome string, store WaitStore, now time.Time) (map[string]any, error) {
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, session)
sw, err := updateWaits(store, key, func(sw SessionWaits) (SessionWaits, error) {
i := findWait(sw, id)
if i < 0 {
return sw, fmt.Errorf("no open wait %s for session %s", id, session)
key := waitsKey(p, name)
nw, err := updateNamedWaits(store, key, func(nw NamedWaits) (NamedWaits, error) {
if err := bindCaller(&nw, name, p.Node, by, session, now); err != nil {
return nw, err
}
sw.Waits = append(append([]Wait(nil), sw.Waits[:i]...), sw.Waits[i+1:]...)
return sw, nil
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 := writeWaitsFile(p, session, sw, now); err != nil {
if err := writeSessionFile(p, session, nw.Waits, now); err != nil {
return nil, fmt.Errorf("the wait was done, but its file could not be written: %w", err)
}
say("wait %s for session %s done: %s", id, session, outcome)
_ = 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, id string, store WaitStore, now time.Time) (map[string]any, error) {
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, session)
sw, err := updateWaits(store, key, func(sw SessionWaits) (SessionWaits, error) {
i := findWait(sw, id)
key := waitsKey(p, name)
nw, 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 sw, fmt.Errorf("no open wait %s for session %s", id, session)
return nw, fmt.Errorf("no open wait %s for %s", id, name)
}
if !sw.Waits[i].PastBound(now) {
return sw, fmt.Errorf("wait %s is not past its bound yet: said is for telling the operator about one that is", id)
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)
}
sw.Waits[i].Said = true
return sw, nil
nw.Waits[i].Said = true
return nw, nil
})
if err != nil {
return nil, err
}
if err := writeWaitsFile(p, session, sw, now); err != nil {
if err := writeSessionFile(p, session, nw.Waits, 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 session's open waits, with their age and whether they are past their bound (1629/AC1).
func ListWaits(p Paths, session string, store WaitStore, now time.Time) (map[string]any, error) {
// ListWaits answers a name's open waits, with their age and whether they are past their bound (1629/AC1).
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
}
sw, _, _, err := readSessionWaits(store, waitsKey(p, session))
name, err := callerName(p, by)
if err != nil {
return nil, fmt.Errorf("the wait list could not be read: %w", err)
return nil, err
}
key := waitsKey(p, name)
nw, err := updateNamedWaits(store, key, func(nw NamedWaits) (NamedWaits, error) {
if err := bindCaller(&nw, name, p.Node, by, session, now); err != nil {
return nw, err
}
return nw, nil
})
if err != nil {
return nil, err
}
out := []map[string]any{}
for _, w := range sw.Waits {
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}
@@ -422,13 +553,63 @@ func ListWaits(p Paths, session string, store WaitStore, now time.Time) (map[str
}
out = append(out, entry)
}
return map[string]any{"session": session, "open": out}, nil
return map[string]any{"session": session, "name": name, "open": out}, 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 {
if err := writeSessionFile(p, oldSession, nil, 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 := writeSessionFile(p, session, nw.Waits, 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)
}
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) {
@@ -443,27 +624,46 @@ func readWaitsFile(path string) (WaitsFile, error) {
return f, nil
}
// waitTools is claude_code_wait: add, watched, done, said and list, over one session's wait list (novox/hq
// ADR 0489).
// 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 session's wait list (novox/hq ADR 0489): 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 or list; session " +
"is always required (the id the SessionStart hook named, or printed again after a /clear). add takes by " +
"(your session's name, as who@node or who), 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 session. 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.",
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 or list"),
"session": str("required: the session's id"),
"by": str("add: your session's name, as who@node or who"),
"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{
@@ -480,7 +680,7 @@ func waitTools(p Paths) []stdio.Tool {
if waiting == nil {
return nil, notLaunched
}
session := strArg(a, "session")
session, by := strArg(a, "session"), strArg(a, "by")
now := time.Now()
if waiting.Now != nil {
now = waiting.Now()
@@ -498,18 +698,20 @@ func waitTools(p Paths) []stdio.Tool {
if n, ok := a["bound_minutes"].(float64); ok {
bound = int(n)
}
return AddWait(p, WaitInput{Session: session, By: strArg(a, "by"), What: strArg(a, "what"), Read: read,
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, strArg(a, "id"), strArg(a, "watcher"), waiting.Store, now)
return WatchedWait(p, session, by, strArg(a, "id"), strArg(a, "watcher"), waiting.Store, now)
case "done":
return DoneWait(p, session, strArg(a, "id"), strArg(a, "outcome"), waiting.Store, now)
return DoneWait(p, session, by, strArg(a, "id"), strArg(a, "outcome"), waiting.Store, now)
case "said":
return SaidWait(p, session, strArg(a, "id"), waiting.Store, now)
return SaidWait(p, session, by, strArg(a, "id"), waiting.Store, now)
case "list":
return ListWaits(p, session, waiting.Store, now)
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 or list, not %q", op)
return nil, fmt.Errorf("op is add, watched, done, said, list or adopt, not %q", op)
}
}},
}