jschoubben
2927cb2ceb
A delivery's own restarts reached the operator as notifications while it ran, and every restart of the tool runner popped the four silenced messages up again on the desk, twice. The router now reads the controller's walk start and end on plan-moved, holds what an open walk touches until its end or bound, tells the end as one message, and makes the restart pass once per restart from its own state, leaving silenced, withheld, answered and held messages as they are.
709 lines
25 KiB
Go
709 lines
25 KiB
Go
// messenger: the holder of the operator-channel seat, the router (novox/hq to-be 45 §5, to-be 46, ADR
|
|
// 0227, ADR 0259). A Go bundle the node's runtime launches. It consumes the controller's condition events and
|
|
// decides what is said to the operator, on the channel kinds the mesh holds — each a module of its own — and
|
|
// holds every ask, judges each answer and issues the warrant its asker acts on. It keeps its open messages in its own state, so a restart forgets nothing. stdout is the MCP
|
|
// channel; what this module says, it says on stderr.
|
|
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
_ "time/tzdata" // the operator's zone is known whatever this machine has installed
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
)
|
|
|
|
func errorf(format string, a ...any) error { return fmt.Errorf(format, a...) }
|
|
|
|
func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }
|
|
|
|
// Settings are the operator's values for this module, merged by the mesh into one JSON file.
|
|
type Settings struct {
|
|
// TimeZone is the operator's, an IANA name such as Europe/Brussels: every time a message says is said
|
|
// in it. Empty is this machine's own zone.
|
|
TimeZone string `json:"time-zone"`
|
|
// OnlyWhatNeedsMe are the channel kinds set to say only what needs the operator: a walk that delivered
|
|
// with nothing held and nothing open is not told there at all (novox/hq ADR 0308 §4).
|
|
OnlyWhatNeedsMe []string `json:"only-what-needs-me"`
|
|
}
|
|
|
|
// zoneOf is the location a time-zone setting names, or this machine's own when it names none or one
|
|
// that does not exist (said, never guessed silently).
|
|
func zoneOf(name string) (*time.Location, error) {
|
|
if name == "" {
|
|
return time.Local, nil
|
|
}
|
|
loc, err := time.LoadLocation(name)
|
|
if err != nil {
|
|
return time.Local, fmt.Errorf("the time-zone %q is not one this machine knows; times are said in its own zone", name)
|
|
}
|
|
return loc, nil
|
|
}
|
|
|
|
func readSettings(path string) (Settings, error) {
|
|
var s Settings
|
|
if path == "" {
|
|
return s, errors.New("started without a settings file")
|
|
}
|
|
raw, err := os.ReadFile(path)
|
|
if err != nil {
|
|
return s, err
|
|
}
|
|
var loose map[string]any
|
|
if err := json.Unmarshal(raw, &loose); err != nil {
|
|
return s, fmt.Errorf("the settings file is not JSON: %v", err)
|
|
}
|
|
if tz, ok := loose["time-zone"].(string); ok {
|
|
s.TimeZone = strings.TrimSpace(tz)
|
|
}
|
|
switch v := loose["only-what-needs-me"].(type) {
|
|
case nil:
|
|
case []any:
|
|
for _, k := range v {
|
|
kind, ok := k.(string)
|
|
if !ok || strings.TrimSpace(kind) == "" {
|
|
return s, fmt.Errorf("only-what-needs-me lists something that is not a channel kind: %v", k)
|
|
}
|
|
s.OnlyWhatNeedsMe = append(s.OnlyWhatNeedsMe, strings.TrimSpace(kind))
|
|
}
|
|
default:
|
|
return s, fmt.Errorf("only-what-needs-me is not a list of channel kinds")
|
|
}
|
|
return s, nil
|
|
}
|
|
|
|
// stateStore keeps the open messages in the module's declared state `open`, and the recent sends
|
|
// under one key of `sent` (ADR 0201).
|
|
type stateStore struct{}
|
|
|
|
// kvKey is a condition key as a bucket key: only the characters a key may carry.
|
|
func kvKey(key string) string {
|
|
var b strings.Builder
|
|
for _, r := range key {
|
|
switch {
|
|
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_', r == '.', r == '=':
|
|
b.WriteRune(r)
|
|
default:
|
|
b.WriteRune('_')
|
|
}
|
|
}
|
|
return strings.Trim(b.String(), ".")
|
|
}
|
|
|
|
func (stateStore) Put(r Record) error {
|
|
_, err := stdio.State("open").Put(kvKey(r.Key), r)
|
|
return err
|
|
}
|
|
func (stateStore) Delete(key string) error { return stdio.State("open").Delete(kvKey(key)) }
|
|
func (stateStore) All() ([]Record, error) {
|
|
keys, err := stdio.State("open").Keys()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var out []Record
|
|
for _, k := range keys {
|
|
e, err := stdio.State("open").Get(k)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if e == nil {
|
|
continue
|
|
}
|
|
var r Record
|
|
if err := json.Unmarshal(e.Value, &r); err != nil {
|
|
logf("[messenger] the open message kept as %s cannot be read (%v); left as it is", k, err)
|
|
continue
|
|
}
|
|
out = append(out, r)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// The walks open now, one key each in the state `walks`, and the restart passes under one key of `sent`
|
|
// (novox/hq ADR 0308): a router that starts again holds, tells and passes as if it had not.
|
|
func (stateStore) PutWalk(w Walk) error {
|
|
_, err := stdio.State("walks").Put(kvKey(w.ID), w)
|
|
return err
|
|
}
|
|
func (stateStore) DeleteWalk(id string) error { return stdio.State("walks").Delete(kvKey(id)) }
|
|
func (stateStore) Walks() ([]Walk, error) {
|
|
keys, err := stdio.State("walks").Keys()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var out []Walk
|
|
for _, k := range keys {
|
|
e, err := stdio.State("walks").Get(k)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if e == nil {
|
|
continue
|
|
}
|
|
var w Walk
|
|
if err := json.Unmarshal(e.Value, &w); err != nil {
|
|
logf("[messenger] the walk kept as %s cannot be read (%v); left as it is", k, err)
|
|
continue
|
|
}
|
|
out = append(out, w)
|
|
}
|
|
return out, nil
|
|
}
|
|
func (stateStore) PutPasses(p map[string]Pass) error {
|
|
_, err := stdio.State("sent").Put("passes", p)
|
|
return err
|
|
}
|
|
func (stateStore) Passes() (map[string]Pass, error) {
|
|
e, err := stdio.State("sent").Get("passes")
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
var p map[string]Pass
|
|
return p, json.Unmarshal(e.Value, &p)
|
|
}
|
|
|
|
func (stateStore) PutRecent(s []Sent) error {
|
|
_, err := stdio.State("sent").Put("recent", s)
|
|
return err
|
|
}
|
|
func (stateStore) Recent() ([]Sent, error) {
|
|
e, err := stdio.State("sent").Get("recent")
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
var s []Sent
|
|
return s, json.Unmarshal(e.Value, &s)
|
|
}
|
|
|
|
// askStore keeps the asks in the state `asks` and the operator's identities, and the links asked for, in
|
|
// `identities` (novox/hq ADR 0259). Each write that must win once is a compare-and-set.
|
|
type askStore struct{}
|
|
|
|
// changed reads the runtime's refusal of a compare-and-set that lost, told by its code, as ErrChanged.
|
|
func changed(err error) error {
|
|
if errors.Is(err, stdio.ErrChanged) {
|
|
return ErrChanged
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (askStore) GetAsk(key string) (*AskRecord, uint64, error) {
|
|
e, err := stdio.State("asks").Get(key)
|
|
if err != nil || e == nil {
|
|
return nil, 0, err
|
|
}
|
|
var r AskRecord
|
|
if err := json.Unmarshal(e.Value, &r); err != nil {
|
|
return nil, 0, fmt.Errorf("the ask kept as %s cannot be read: %v", key, err)
|
|
}
|
|
return &r, e.Revision, nil
|
|
}
|
|
func (askStore) CreateAsk(key string, r AskRecord) (uint64, error) {
|
|
rev, err := stdio.State("asks").Create(key, r)
|
|
return rev, changed(err)
|
|
}
|
|
func (askStore) UpdateAsk(key string, r AskRecord, revision uint64) (uint64, error) {
|
|
rev, err := stdio.State("asks").Update(key, r, revision)
|
|
return rev, changed(err)
|
|
}
|
|
func (askStore) AskKeys() ([]string, error) { return stdio.State("asks").Keys() }
|
|
func (askStore) GetIdentity(key string) (*Identity, error) {
|
|
e, err := stdio.State("identities").Get(key)
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
var id Identity
|
|
return &id, json.Unmarshal(e.Value, &id)
|
|
}
|
|
func (askStore) PutIdentity(key string, id Identity) error {
|
|
_, err := stdio.State("identities").Put(key, id)
|
|
return err
|
|
}
|
|
func (askStore) DeleteIdentity(key string) error { return stdio.State("identities").Delete(key) }
|
|
func (askStore) IdentityKeys() ([]string, error) {
|
|
keys, err := stdio.State("identities").Keys()
|
|
var out []string
|
|
for _, k := range keys {
|
|
if !strings.HasPrefix(k, "pending.") && !strings.HasPrefix(k, "unlinked.") && !strings.HasPrefix(k, "asked.") {
|
|
out = append(out, k)
|
|
}
|
|
}
|
|
return out, err
|
|
}
|
|
func (askStore) GetLink(kind string) (*PendingLink, error) {
|
|
e, err := stdio.State("identities").Get("pending." + kind)
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
var l PendingLink
|
|
return &l, json.Unmarshal(e.Value, &l)
|
|
}
|
|
func (askStore) PutLink(kind string, l PendingLink) error {
|
|
_, err := stdio.State("identities").Put("pending."+kind, l)
|
|
return err
|
|
}
|
|
func (askStore) DeleteLink(kind string) error {
|
|
return stdio.State("identities").Delete("pending." + kind)
|
|
}
|
|
func (askStore) GetUnlinked(kind string) (*Unlinked, error) {
|
|
e, err := stdio.State("identities").Get("unlinked." + kind)
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
var u Unlinked
|
|
return &u, json.Unmarshal(e.Value, &u)
|
|
}
|
|
func (askStore) PutUnlinked(kind string, u Unlinked) error {
|
|
_, err := stdio.State("identities").Put("unlinked."+kind, u)
|
|
return err
|
|
}
|
|
|
|
func (askStore) GetLinksAsked(key string) ([]time.Time, error) {
|
|
e, err := stdio.State("identities").Get(key)
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
var times []time.Time
|
|
return times, json.Unmarshal(e.Value, ×)
|
|
}
|
|
func (askStore) PutLinksAsked(key string, times []time.Time) error {
|
|
_, err := stdio.State("identities").Put(key, times)
|
|
return err
|
|
}
|
|
|
|
// listening is whether condition events reach this holder, in words.
|
|
type listening struct {
|
|
mu sync.Mutex
|
|
now string
|
|
}
|
|
|
|
func (l *listening) set(s string) { l.mu.Lock(); l.now = s; l.mu.Unlock() }
|
|
func (l *listening) get() string { l.mu.Lock(); defer l.mu.Unlock(); return l.now }
|
|
|
|
func main() {
|
|
settingsFile := os.Getenv("MESH_MESSENGER_SETTINGS")
|
|
var said sync.Mutex
|
|
lastSaid, zoneSaid := "", ""
|
|
settings := func() Settings {
|
|
s, err := readSettings(settingsFile)
|
|
said.Lock()
|
|
defer said.Unlock()
|
|
if err != nil && err.Error() != lastSaid {
|
|
logf("[messenger] settings cannot be read: %v", err)
|
|
}
|
|
lastSaid = ""
|
|
if err != nil {
|
|
lastSaid = err.Error()
|
|
}
|
|
return s
|
|
}
|
|
bench := NewBench(time.Now)
|
|
publish := func(subject string, body any, id string) error { return stdio.SeatPublish(subject, body, id) }
|
|
h := &Holder{
|
|
Away: &SeatChannel{Bench: bench, Role: RoleAway, Publish: publish},
|
|
Desk: &SeatChannel{Bench: bench, Role: RoleDesk, Publish: publish},
|
|
Store: stateStore{},
|
|
Asks: askStore{},
|
|
Node: os.Getenv("MESH_NODE"),
|
|
StateDir: os.Getenv("MESH_MESSENGER_STATE"),
|
|
Bench: bench,
|
|
Publish: publish,
|
|
Ask: stdio.Ask,
|
|
Logf: logf,
|
|
Zone: func() *time.Location {
|
|
loc, err := zoneOf(settings().TimeZone)
|
|
said.Lock()
|
|
defer said.Unlock()
|
|
if err != nil && err.Error() != zoneSaid {
|
|
logf("[messenger] %v", err)
|
|
}
|
|
zoneSaid = ""
|
|
if err != nil {
|
|
zoneSaid = err.Error()
|
|
}
|
|
return loc
|
|
},
|
|
Emit: func(event string, body any) error { return stdio.Emit(event, body) },
|
|
NeedsOnly: func(kind string) bool {
|
|
for _, k := range settings().OnlyWhatNeedsMe {
|
|
if k == kind {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
},
|
|
}
|
|
bench.To = func(kind string) string {
|
|
if ids := h.identitiesOf(kind); len(ids) > 0 {
|
|
return ids[0].Identity
|
|
}
|
|
return ""
|
|
}
|
|
h.init()
|
|
l := &listening{now: "not yet: starting"}
|
|
go run(h, l)
|
|
if err := stdio.Serve("", tools(h, l)); err != nil {
|
|
logf("%v", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
// run reads back the state, listens for condition events, and keeps time — each retried, and each
|
|
// failure said, never given up on quietly.
|
|
func run(h *Holder, l *listening) {
|
|
time.Sleep(500 * time.Millisecond) // Serve first: the state is reached through it
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
err := h.Load()
|
|
if err == nil {
|
|
break
|
|
}
|
|
logf("[messenger] cannot read back its open messages yet (%v); asking again in %s", err, wait)
|
|
time.Sleep(wait)
|
|
}
|
|
for _, ch := range h.channels() {
|
|
if err := ch.Ready(); err != nil {
|
|
logf("[messenger] %s cannot send: %v", ch.Name(), err)
|
|
}
|
|
}
|
|
// The state first: what is open now is read from the controller before any event is taken, so
|
|
// what the stream replays is history (novox/hq issue 271). Not reachable yet, it is asked for
|
|
// again in the loop below, and old events meanwhile are kept as state, never said.
|
|
h.Start()
|
|
syncFromController(h)
|
|
go func() {
|
|
tick := time.NewTicker(5 * time.Second)
|
|
defer tick.Stop()
|
|
lastTick, lastAsk := time.Now(), time.Now()
|
|
for now := range tick.C {
|
|
h.Flush()
|
|
if now.Sub(lastTick) >= time.Minute {
|
|
lastTick = now
|
|
h.Tick()
|
|
}
|
|
if h.WantsSync() && now.Sub(lastAsk) >= time.Minute {
|
|
lastAsk = now
|
|
syncFromController(h)
|
|
}
|
|
}
|
|
}()
|
|
handle := func(e stdio.Envelope) error {
|
|
if e.Key == ControllerSeat+"."+EventPlanMoved || e.Key == ControllerSeat+"."+EventWalk {
|
|
// A walk's start or end (novox/hq ADR 0308 §1); any other move of a plan is not read here.
|
|
ev, isWalk, err := DecodeWalk(e.Body)
|
|
if err != nil {
|
|
h.Unreadable(e.Key, err)
|
|
return nil
|
|
}
|
|
if isWalk {
|
|
h.Walked(ev)
|
|
}
|
|
return nil
|
|
}
|
|
event, ok := eventOf(e.Key)
|
|
if !ok {
|
|
return nil
|
|
}
|
|
c, err := DecodeCondition(event, e.Body)
|
|
if err != nil {
|
|
h.Unreadable(e.Key, err)
|
|
return nil
|
|
}
|
|
h.Condition(event, c)
|
|
return nil
|
|
}
|
|
go listenForChannels(h)
|
|
go takeAsks(h)
|
|
go answerProofs(h)
|
|
go followKinds(h)
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
err := stdio.Subscribe(ControllerSeat+".*", handle)
|
|
if err == nil {
|
|
l.set("listening")
|
|
logf("[messenger] listening for the controller's condition events")
|
|
return
|
|
}
|
|
l.set("not yet: " + err.Error())
|
|
logf("[messenger] not hearing condition events yet (%v); asking again in %s", err, wait)
|
|
time.Sleep(wait)
|
|
}
|
|
}
|
|
|
|
// kindOf is an intake event's kind and what it is: `intake.<what>.<kind>`, the kind the subject's last token,
|
|
// which the bus grants that kind's holder alone.
|
|
func kindOf(key string) (what, kind string, ok bool) {
|
|
parts := strings.Split(key, ".")
|
|
if len(parts) != 3 || parts[0] != asks.IntakeSeat {
|
|
return "", "", false
|
|
}
|
|
return parts[1], parts[2], true
|
|
}
|
|
|
|
// listenForChannels takes what the channels say (novox/hq ADR 0259): answers chosen, links asked, messages
|
|
// that failed, and whether each can send — every one under the kind the bus granted its holder.
|
|
func listenForChannels(h *Holder) {
|
|
handle := func(e stdio.Envelope) error {
|
|
what, kind, ok := kindOf(e.Key)
|
|
if !ok {
|
|
return nil
|
|
}
|
|
switch what {
|
|
case asks.Chosen:
|
|
var c asks.Choice
|
|
if err := json.Unmarshal(e.Body, &c); err != nil || c.Ticket == "" {
|
|
logf("[messenger] an answer from %s could not be read: %s", kind, firstLineOf(string(e.Body)))
|
|
return nil
|
|
}
|
|
if c.ID == "" {
|
|
c.ID = e.Headers["x-event-id"]
|
|
}
|
|
_, _ = h.Chosen(kind, c)
|
|
case asks.Link:
|
|
l, err := decodeLink(e.Body)
|
|
if err != nil {
|
|
logf("[messenger] a link asked on %s could not be read", kind)
|
|
return nil
|
|
}
|
|
h.Link(kind, l)
|
|
case asks.FailedWhat:
|
|
var f asks.Failed
|
|
if json.Unmarshal(e.Body, &f) == nil {
|
|
h.Failed(kind, f)
|
|
}
|
|
case asks.StandingWhat:
|
|
// Read as this module's Standing: the bench's word, and answers-lost-at beside it (issue 384).
|
|
var s Standing
|
|
if json.Unmarshal(e.Body, &s) == nil {
|
|
h.Stand(kind, s)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
if err := stdio.Subscribe(asks.IntakeSeat+".#", handle); err == nil {
|
|
logf("[messenger] listening for what the channels say")
|
|
break
|
|
} else {
|
|
logf("[messenger] not hearing the channels yet (%v); asking again in %s", err, wait)
|
|
}
|
|
time.Sleep(wait)
|
|
}
|
|
}
|
|
|
|
// takeAsks takes the asks and cancels submitted to the seat, each under its asker's name.
|
|
func takeAsks(h *Holder) {
|
|
handle := func(w stdio.Work) error {
|
|
parts := strings.Split(w.Subject, ".")
|
|
if len(parts) != 6 {
|
|
logf("[messenger] work on %s is not an ask; taken and dropped", w.Subject)
|
|
return nil
|
|
}
|
|
return h.TakeAsk(parts[4], parts[5], w.Body)
|
|
}
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
if err := stdio.SeatTake(AsksWorker, handle); err == nil {
|
|
logf("[messenger] taking the asks submitted to %s", asks.Seat)
|
|
break
|
|
} else {
|
|
logf("[messenger] not taking asks yet (%v); asking again in %s", err, wait)
|
|
}
|
|
time.Sleep(wait)
|
|
}
|
|
}
|
|
|
|
// AsksWorker is the worker the seat's asks are taken from, as the controller names it.
|
|
const AsksWorker = "SEAT_OPERATOR_CHANNEL_worker"
|
|
|
|
// answerProofs answers the codes the channels ask as proofs: never an event, never kept.
|
|
func answerProofs(h *Holder) {
|
|
handle := func(subject string, body json.RawMessage) (any, error) {
|
|
parts := strings.Split(subject, ".")
|
|
kind := parts[len(parts)-1]
|
|
var c asks.Code
|
|
if err := json.Unmarshal(body, &c); err != nil {
|
|
return asks.ProofAnswer{Words: "The code could not be read."}, nil
|
|
}
|
|
return h.Prove(kind, c), nil
|
|
}
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
if err := stdio.SeatAnswer(asks.ProofSubject(asks.CodeProof, "*"), handle); err == nil {
|
|
logf("[messenger] answering the codes the channels are sent")
|
|
break
|
|
} else {
|
|
logf("[messenger] not answering codes yet (%v); asking again in %s", err, wait)
|
|
}
|
|
time.Sleep(wait)
|
|
}
|
|
}
|
|
|
|
// followKinds reads, every half minute, which channel kinds the controller says exist and what each
|
|
// promises: never from a channel.
|
|
func followKinds(h *Holder) {
|
|
said := ""
|
|
for {
|
|
kinds, err := stdio.SeatKinds()
|
|
if err != nil {
|
|
if err.Error() != said {
|
|
logf("[messenger] which channels exist cannot be read (%v); asking again", err)
|
|
said = err.Error()
|
|
}
|
|
} else {
|
|
var out []KindInfo
|
|
for _, k := range kinds {
|
|
if k.Seat == asks.ChannelSeat {
|
|
out = append(out, KindInfo{Kind: k.Kind, Module: k.Module, Node: k.Node, Capabilities: k.Capabilities})
|
|
}
|
|
}
|
|
h.Bench.SetKinds(out)
|
|
said = ""
|
|
}
|
|
time.Sleep(30 * time.Second)
|
|
}
|
|
}
|
|
|
|
// syncFromController reads the controller's open conditions and gives them to the holder.
|
|
func syncFromController(h *Holder) {
|
|
asked := h.Now()
|
|
type answer struct {
|
|
raw json.RawMessage
|
|
err error
|
|
}
|
|
done := make(chan answer, 1)
|
|
go func() {
|
|
raw, err := stdio.Ask("seat:"+ControllerSeat+".conditions", map[string]any{})
|
|
done <- answer{raw, err}
|
|
}()
|
|
var a answer
|
|
select {
|
|
case a = <-done:
|
|
case <-time.After(30 * time.Second):
|
|
a.err = errors.New("no answer in 30 s")
|
|
}
|
|
if a.err != nil {
|
|
h.SyncFailed(a.err)
|
|
return
|
|
}
|
|
open, err := DecodeOpen(a.raw)
|
|
if err != nil {
|
|
h.SyncFailed(err)
|
|
return
|
|
}
|
|
h.Sync(open, asked)
|
|
}
|
|
|
|
func str(description string) map[string]any {
|
|
return map[string]any{"type": "string", "description": description}
|
|
}
|
|
|
|
func strArg(a map[string]any, k string) string { s, _ := a[k].(string); return strings.TrimSpace(s) }
|
|
|
|
func limitArg(a map[string]any, def int) int {
|
|
if v, ok := a["limit"].(float64); ok && v >= 1 {
|
|
return min(int(v), KeptSends)
|
|
}
|
|
return def
|
|
}
|
|
|
|
func tools(h *Holder, l *listening) []stdio.Tool {
|
|
return []stdio.Tool{
|
|
{Name: "operator-channel.open",
|
|
Description: "What is open now: every message the operator was sent about something still wrong, urgent first, " +
|
|
"oldest first — its key, severity, summary, since when, the channels it went to, whether it is silenced, " +
|
|
"reminded, held (for its burst window, the desktop's rhythm or the cap), refused, not sent yet, or " +
|
|
"never said (known from the state or an old event: history, not news).",
|
|
Run: func(map[string]any) (any, error) { return h.Open(), nil }},
|
|
{Name: "operator-channel.history",
|
|
Description: "What was said to the operator lately, newest first — each send, digest, edit and failure with its " +
|
|
"channel, key and outcome — and every message refused for carrying an address, a path or a secret.",
|
|
Input: map[string]any{"limit": map[string]any{"type": "integer", "description": "at most this many of each (default 50)"}},
|
|
Run: func(a map[string]any) (any, error) { return h.History(limitArg(a, 50)), nil }},
|
|
{Name: "operator-channel.notify",
|
|
Description: "Tell the operator something, as a module that uses the seat: a key (the same key is the same " +
|
|
"message), urgent or warning, one line in the mesh's words, and what it is about; and best, in plain " +
|
|
"words, a headline, an explanation and the line said when it is over. clear says it is over. " +
|
|
"Roles and words only: an address, a path or a secret is refused. Deduplicated, capped and routed " +
|
|
"like the controller's conditions.",
|
|
Input: map[string]any{
|
|
"key": str("what makes it the same message the next time, e.g. backup.ace.failed"),
|
|
"severity": map[string]any{"type": "string", "enum": []string{Urgent, Warning}},
|
|
"summary": str("one line in the mesh's words"),
|
|
"subject": str("what it is about: a machine's role, a module, a plan"),
|
|
"headline": str("a few plain words the operator reads as the title, e.g. backup failed on ace: no ids, keys or commands"),
|
|
"explanation": str("one or two plain sentences: what happened, what it means for the operator, and whether " +
|
|
"they need to act"),
|
|
"resolved": str("the one short line said when it is over, e.g. backups run again on ace"),
|
|
"clear": map[string]any{"type": "boolean", "description": "it is over: the message is edited to say so"},
|
|
},
|
|
Run: func(a map[string]any) (any, error) {
|
|
clear, _ := a["clear"].(bool)
|
|
return h.Notify(strArg(a, "key"), strArg(a, "severity"), strArg(a, "summary"), strArg(a, "subject"), clear,
|
|
strArg(a, "headline"), strArg(a, "explanation"), strArg(a, "resolved"))
|
|
}},
|
|
{Name: "messenger_status",
|
|
Description: "Whether the operator can be reached, and why not: each channel kind the mesh holds — can it " +
|
|
"send, what it said of itself, its last error, how many it sent in the last hour against the cap, how " +
|
|
"many it holds — the open and silenced count, messages not sent yet, refusals, unreadable events, " +
|
|
"whether condition events arrive, and the rules it applies.",
|
|
Run: func(map[string]any) (any, error) { return h.Status(l.get()), nil }},
|
|
{Name: "operator-channel.asks",
|
|
Description: "The asks this router holds (novox/hq ADR 0259): open first, then the last month's — each " +
|
|
"one's asker, headline, state, when it opened and expires, the channels it was offered on, how it " +
|
|
"ended (the warrant: who chose what, through which channel), and whether its asker was told.",
|
|
Run: func(map[string]any) (any, error) { return h.AsksNow(), nil }},
|
|
{Name: "messenger_identities",
|
|
Description: "The accounts linked as the operator, one per channel kind: the account, the name its " +
|
|
"service shows, when it was linked, and from when it can approve (an hour after linking).",
|
|
Run: func(map[string]any) (any, error) { return h.Identities(), nil }},
|
|
{Name: "messenger_unlink",
|
|
Description: "Unlink the account linked as the operator on one channel kind: from then its answers are " +
|
|
"refused until an account is linked again. Said on every channel, the unlinked one included.",
|
|
Input: map[string]any{"kind": str("the channel kind, e.g. telegram")},
|
|
Run: func(a map[string]any) (any, error) {
|
|
kind := strArg(a, "kind")
|
|
if kind == "" {
|
|
return nil, errorf("kind is required")
|
|
}
|
|
return h.Unlink(kind)
|
|
}},
|
|
{Name: "messenger_recent",
|
|
Description: "The recent sends — each message, digest, edit and failure on each channel, newest first — and the refusals.",
|
|
Input: map[string]any{"limit": map[string]any{"type": "integer", "description": "at most this many (default 20)"}},
|
|
Run: func(a map[string]any) (any, error) { return h.History(limitArg(a, 20)), nil }},
|
|
{Name: "messenger_test",
|
|
Description: "Send a test message now, to one channel kind or every one (default), through the content rule " +
|
|
"and the cap: proves a channel reaches the operator. Answers per channel: put on its queue, or why not.",
|
|
Input: map[string]any{
|
|
"channel": str("a channel kind, e.g. telegram, or empty for every one"),
|
|
"text": str("the words (default: a line saying it is a test)"),
|
|
},
|
|
Run: func(a map[string]any) (any, error) { return h.Test(strArg(a, "channel"), strArg(a, "text")), nil }},
|
|
{Name: "messenger_preview",
|
|
Description: "What a condition's message would say, without sending it: the title, the words, the " +
|
|
"line on the issues that track it, and whether the content rule would let them leave the mesh — " +
|
|
"where they may not, the reduced message a send would carry instead. Composed for a condition this " +
|
|
"router holds open, by the walk a send composes with; a clearing as if it cleared now, a silence " +
|
|
"only while there is one. It sends, queues and records nothing.",
|
|
Input: map[string]any{
|
|
"key": str("the condition's key, as the controller names it, e.g. module.openrazer.laptop.relogin-needed"),
|
|
"what": str("which message about it: raised (default), reminder, quiet-reminder, escalated, reopened, cleared, silenced"),
|
|
},
|
|
Run: func(a map[string]any) (any, error) { return h.Preview(strArg(a, "key"), strArg(a, "what")) }},
|
|
{Name: "messenger_check",
|
|
Description: "Would these words be allowed to leave the mesh? Runs the content rule — no address, path or " +
|
|
"secret — and answers allowed, or the class of what it carried. Sends nothing.",
|
|
Input: map[string]any{"text": str("the words to judge")},
|
|
Run: func(a map[string]any) (any, error) {
|
|
if r, ok := Check(strArg(a, "text")); !ok {
|
|
return map[string]any{"allowed": false, "class": r.Class, "carried": r.What}, nil
|
|
}
|
|
return map[string]any{"allowed": true}, nil
|
|
}},
|
|
}
|
|
}
|