diff --git a/README.md b/README.md index 01d1d3e..45b8607 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,21 @@ runtime's (`node-tools/internal/launch`, mesh-tools). values and then every change. Through the runtime that launched the bundle, which refuses what the module did not declare. Current values, never history and never a secret (novox/hq ADR 0201). +### `seats` and `asks` — asking a person, answering on a channel (Go) +A seat may name its caller or its kind (novox/hq ADR 0259): `stdio.SeatPublish` submits an accept or says +an event under the module's own name or kind, `stdio.SeatTake` takes a held seat's work, `stdio.SeatProve` +and `stdio.SeatAnswer` carry a proof (a code typed by a person, by request and reply, never kept), +`stdio.SeatRecord` reads a record kept for the module, and `State(...).Create` and `.Update` write by +compare-and-set. The runtime does each only as far as the module's membership lists. `go/asks` is the +contract on top: the `Ask` an asker publishes and the checks it is held to, the `Warrant` it acts on +(`For`, `Says`), and the `Message`, `Choice`, `LinkAsked` and `Code` a channel takes and says, with the +subjects each travels on. +`go/asker` is an asking module's client (ADR 0259 §10): `Ask` keeps an ask in the module's own state and then +publishes it under the module's name; `Cancel`; `Where` reads its state from the router's record; `Listen` and +`Take` hand the module each warrant once — marked acted on by compare-and-set before the module acts, and only +for an ask it made, an option it offered, at that option's level, before it expired. A lost compare-and-set is +`stdio.ErrChanged`, told by the runtime's error code. + ### `primitives` — sealing, semver, resolved-env Seal/unseal for secrets, semantic version comparison, and the accessor a module uses to read its own resolved environment. diff --git a/go/asker/asker.go b/go/asker/asker.go new file mode 100644 index 0000000..360c7b6 --- /dev/null +++ b/go/asker/asker.go @@ -0,0 +1,196 @@ +// Package asker is a module's client for asking a person (novox/hq ADR 0259 §10): it publishes an ask under +// the module's own name, cancels it, reads its state from the router's record, and hands the module each +// warrant once — only one for an ask it made, for an option it offered, at that option's level, before it +// expired. The module then acts with its own grants. +// +// It keeps what it asked, and which warrants it acted on, in a state of the module's own (declared in its +// manifest, by default `asked`): the ask is written before it is published, and a warrant is marked acted on +// by compare-and-set before the module's handler runs, so a redelivery, or a second instance, acts once. +package asker + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" +) + +// Client asks as one module. +type Client struct { + // Module is the asking module's name: the one the bus grants its asks under. + Module string + // State is the module's own state the asks are kept in. + State Store + // Now is the clock; nil is time.Now. + Now func() time.Time + // Publish, Record and Subscribe reach the bus; nil is the stdio SDK's, through the module's runtime. + Publish func(subject string, body any, id string) error + Record func(key string) (*stdio.StateEntry, error) + Subscribe func(pattern string, handle func(stdio.Envelope) error) error +} + +// Store is where the asks are kept: stdio.State(name) is one. +type Store interface { + Get(key string) (*stdio.StateEntry, error) + Create(key string, value any) (uint64, error) + Update(key string, value any, revision uint64) (uint64, error) +} + +// New is a client for the module the runtime serves, keeping its asks in the state named. +func New(state string) *Client { + return &Client{Module: os.Getenv("MESH_MODULE"), State: stdio.State(state)} +} + +func (c *Client) now() time.Time { + if c.Now != nil { + return c.Now() + } + return time.Now() +} + +// Kept is one ask as the module keeps it. +type Kept struct { + Ask asks.Ask `json:"ask"` + Asked time.Time `json:"asked"` + Acted bool `json:"acted,omitempty"` + Ended asks.Outcome `json:"ended,omitempty"` + Answer *asks.Warrant `json:"answer,omitempty"` +} + +func key(id string) string { return "ask." + id } + +// Ask checks an ask, keeps it, and publishes it under the module's own name. An ask with an id already kept +// is refused: a new question is a new id. +func (c *Client) Ask(a asks.Ask) error { + if err := a.Check(c.now()); err != nil { + return err + } + if _, err := c.State.Create(key(a.ID), Kept{Ask: a, Asked: c.now()}); err != nil { + if errors.Is(err, stdio.ErrChanged) { + return fmt.Errorf("an ask %s was already made: a new question is a new id", a.ID) + } + return err + } + return c.publish(asks.AskSubject(c.Module), a, a.ID) +} + +// Cancel takes an ask back: the router ends it and edits every copy. +func (c *Client) Cancel(id string) error { + return c.publish(asks.CancelSubject(c.Module), map[string]string{"id": id}, "cancel."+id) +} + +func (c *Client) publish(subject string, body any, id string) error { + if c.Publish != nil { + return c.Publish(subject, body, id) + } + return stdio.SeatPublish(subject, body, id) +} + +// Where is an ask's state as the router keeps it: its state and its warrant, if any — read under the +// module's own name and no other's. +func (c *Client) Where(id string) (state string, w *asks.Warrant, err error) { + read := c.Record + if read == nil { + read = func(k string) (*stdio.StateEntry, error) { return stdio.SeatRecord("", k) } + } + e, err := read(id) + if err != nil || e == nil { + return "", nil, err + } + var rec struct { + State string `json:"state"` + Warrant *asks.Warrant `json:"warrant"` + } + if err := json.Unmarshal(e.Value, &rec); err != nil { + return "", nil, fmt.Errorf("the router's record of %s cannot be read: %w", id, err) + } + return rec.State, rec.Warrant, nil +} + +// Handled is what came of a warrant: acted on, refused (and why), or already done. +type Handled string + +const ( + Acted Handled = "acted" + Refused Handled = "refused" + Done Handled = "already acted on" + Recorded Handled = "ended without a choice" +) + +// Take hands act a warrant once, after holding it to the ask it answers. act performs the option with the +// module's own grants; its error is said and the warrant stays acted on (a person's decision is not +// performed twice; the module says the failure in its own record). +func (c *Client) Take(w asks.Warrant, act func(asks.Option, asks.Warrant) error) (Handled, error) { + if w.Asker != c.Module { + return Refused, fmt.Errorf("a warrant for %s reached %s", w.Asker, c.Module) + } + e, err := c.State.Get(key(w.Ask)) + if err != nil { + return "", err + } + if e == nil { + return Refused, fmt.Errorf("%s holds no ask %s", c.Module, w.Ask) + } + var k Kept + if err := json.Unmarshal(e.Value, &k); err != nil { + return "", err + } + if k.Acted || k.Ended != "" { + return Done, nil + } + if w.Outcome != asks.OutcomeChosen { + k.Ended = w.Outcome + k.Answer = &w + if _, err := c.State.Update(key(w.Ask), k, e.Revision); err != nil && !errors.Is(err, stdio.ErrChanged) { + return "", err + } + return Recorded, nil + } + o, err := w.For(c.Module, k.Ask) + if err != nil { + return Refused, err + } + // Marked before acting: of two deliveries, or two instances, only the one whose write stands acts. + k.Acted, k.Ended, k.Answer = true, asks.OutcomeChosen, &w + if _, err := c.State.Update(key(w.Ask), k, e.Revision); err != nil { + if errors.Is(err, stdio.ErrChanged) { + return Done, nil + } + return "", err + } + return Acted, act(o, w) +} + +// Listen hands every warrant for this module to Take, with act. It asks the runtime for the module's +// events matching the router's `decided` subject for this module. +func (c *Client) Listen(act func(asks.Option, asks.Warrant) error, said func(Handled, asks.Warrant, error)) error { + subscribe := c.Subscribe + if subscribe == nil { + subscribe = stdio.Subscribe + } + return subscribe(asks.Seat+".decided."+c.Module, func(e stdio.Envelope) error { + if !strings.HasPrefix(e.Key, asks.Seat+".decided.") { + return nil + } + var w asks.Warrant + if err := json.Unmarshal(e.Body, &w); err != nil { + if said != nil { + said(Refused, w, fmt.Errorf("a warrant could not be read: %w", err)) + } + return nil + } + h, err := c.Take(w, act) + if h == "" && err != nil { + return err // the state could not be reached: offered again + } + if said != nil { + said(h, w, err) + } + return nil + }) +} diff --git a/go/asker/asker_test.go b/go/asker/asker_test.go new file mode 100644 index 0000000..6285876 --- /dev/null +++ b/go/asker/asker_test.go @@ -0,0 +1,141 @@ +package asker + +import ( + "encoding/json" + "strings" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" +) + +type mem struct { + vals map[string]json.RawMessage + revs map[string]uint64 + n uint64 +} + +func (m *mem) Get(k string) (*stdio.StateEntry, error) { + v, ok := m.vals[k] + if !ok { + return nil, nil + } + return &stdio.StateEntry{Key: k, Value: v, Revision: m.revs[k]}, nil +} +func (m *mem) Create(k string, v any) (uint64, error) { + if _, ok := m.vals[k]; ok { + return 0, stdio.ErrChanged + } + return m.put(k, v) +} +func (m *mem) Update(k string, v any, rev uint64) (uint64, error) { + if rev == 0 || m.revs[k] != rev { + return 0, stdio.ErrChanged + } + return m.put(k, v) +} +func (m *mem) put(k string, v any) (uint64, error) { + raw, _ := json.Marshal(v) + m.n++ + m.vals[k], m.revs[k] = raw, m.n + return m.n, nil +} + +var now = time.Date(2026, 10, 8, 14, 0, 0, 0, time.UTC) + +func client() (*Client, map[string][]string) { + sent := map[string][]string{} + c := &Client{Module: "mesh-delivery", State: &mem{vals: map[string]json.RawMessage{}, revs: map[string]uint64{}}, + Now: func() time.Time { return now }, + Publish: func(subject string, body any, id string) error { + sent[subject] = append(sent[subject], id) + return nil + }} + return c, sent +} + +// releaseHQ is the act the ask's one option binds. +var releaseHQ = map[string]string{"verb": "release", "delivery": "hq"} + +func held() asks.Ask { + binds, _ := asks.ActDigest(releaseHQ) + return asks.Ask{ID: "d-1", Headline: "Delivery of hq held", Explanation: "Needs you: release it, or stop it.", + Who: asks.Operator, Expires: now.Add(time.Hour), OnExpiry: "the delivery stays held", + Options: []asks.Option{{ID: "release", Label: "Release", Does: "it goes on", Level: asks.Approve, Binds: binds}}} +} + +func warrant() asks.Warrant { + return asks.Warrant{Ask: "d-1", Asker: "mesh-delivery", Outcome: asks.OutcomeChosen, Option: "release", + Label: "Release", Level: asks.Approve, At: now.Add(time.Minute), AskDigest: held().Digest(), + By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"}} +} + +func TestAnAskIsKeptThenPublishedUnderTheModulesOwnName(t *testing.T) { + c, sent := client() + if err := c.Ask(held()); err != nil { + t.Fatal(err) + } + if got := sent[asks.AskSubject("mesh-delivery")]; len(got) != 1 || got[0] != "d-1" { + t.Fatalf("%v", sent) + } + if err := c.Ask(held()); err == nil || !strings.Contains(err.Error(), "already made") { + t.Errorf("the same id was asked twice: %v", err) + } + bad := held() + bad.Who = "agent" + if err := c.Ask(bad); err == nil { + t.Error("an ask the contract refuses was published") + } +} + +func TestAWarrantIsActedOnOnceAndOnlyForItsOwnAsk(t *testing.T) { + c, _ := client() + _ = c.Ask(held()) + acted := 0 + act := func(o asks.Option, _ asks.Warrant) error { + if err := o.Performs(releaseHQ); err != nil { + t.Errorf("the bound act: %v", err) + } + acted++ + return nil + } + if h, err := c.Take(warrant(), act); h != Acted || err != nil || acted != 1 { + t.Fatalf("%s %v %d", h, err, acted) + } + if h, _ := c.Take(warrant(), act); h != Done || acted != 1 { + t.Errorf("a redelivery was acted on: %s", h) + } + for why, change := range map[string]func(*asks.Warrant){ + "another asker": func(w *asks.Warrant) { w.Asker = "mesh-controller" }, + "an ask not made": func(w *asks.Warrant) { w.Ask = "d-9" }, + "another level": func(w *asks.Warrant) { w.Level = asks.Acknowledge }, + "an option not offered": func(w *asks.Warrant) { w.Option = "delete" }, + "after its expiry": func(w *asks.Warrant) { w.At = now.Add(2 * time.Hour) }, + "another ask's digest": func(w *asks.Warrant) { w.AskDigest = "sha256:00" }, + } { + c2, _ := client() + _ = c2.Ask(held()) + w := warrant() + change(&w) + if h, _ := c2.Take(w, act); h == Acted { + t.Errorf("a warrant from %s was acted on", why) + } + } + if acted != 1 { + t.Errorf("acted %d times", acted) + } +} + +func TestAnAskEndedWithoutAChoiceIsRecordedAndNothingDone(t *testing.T) { + c, _ := client() + _ = c.Ask(held()) + w := warrant() + w.Outcome, w.Option, w.By = asks.OutcomeExpired, "", nil + if h, err := c.Take(w, func(asks.Option, asks.Warrant) error { t.Fatal("acted"); return nil }); h != Recorded || err != nil { + t.Fatalf("%s %v", h, err) + } + if h, _ := c.Take(warrant(), func(asks.Option, asks.Warrant) error { t.Fatal("acted after it ended"); return nil }); h != Done { + t.Errorf("%s", h) + } +} diff --git a/go/asks/asks.go b/go/asks/asks.go new file mode 100644 index 0000000..81737ac --- /dev/null +++ b/go/asks/asks.go @@ -0,0 +1,494 @@ +// Package asks is the contract of asking a person and answering on a channel (novox/hq ADR 0259): the +// shapes an asker, the router and a channel exchange on the bus, and the subjects they travel on. No +// transport and no channel's service: an asker publishes an Ask under its own name and acts on the Warrant +// it hears; the router holds the ask, sends channels a Message, and judges the Choice a channel says; a +// channel shows a Message and says what was chosen and by whom, as its service authenticated it. +package asks + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "regexp" + "sort" + "strconv" + "strings" + "time" +) + +// The seats (novox/hq ADR 0259 §3). +const ( + // Seat is held by the router: an ask and its cancel are its accepts, a warrant its event, each named by + // the asker. + Seat = "operator-channel" + // ChannelSeat is the kinded bench a channel holds to show and say: its accepts carry the kind. + ChannelSeat = "channel" + // IntakeSeat is the kinded bench a channel holds to say what was chosen: its events and proofs carry + // the kind. + IntakeSeat = "intake" +) + +// AskSubject is where an asker publishes an ask, CancelSubject its cancel, and DecidedSubject where it +// hears the warrant, or the ask's end without one. +func AskSubject(asker string) string { return "mesh.seat." + Seat + ".accept.ask." + asker } +func CancelSubject(asker string) string { return "mesh.seat." + Seat + ".accept.cancel." + asker } +func DecidedSubject(asker string) string { return "mesh.seat." + Seat + ".event.decided." + asker } + +// The work a channel takes, on ChannelSubject. +const ( + Show = "show" // a message offering answers + Edit = "edit" // a message shown before, replaced + Send = "send" // a message offering nothing +) + +// ChannelSubject is where the router sends a channel of a kind its work. +func ChannelSubject(verb, kind string) string { + return "mesh.seat." + ChannelSeat + ".accept." + verb + "." + kind +} + +// What a channel says, on IntakeSubject. +const ( + Chosen = "choice" // a button tapped + Link = "link" // somebody asked to be linked as the operator +) + +// IntakeSubject is where a channel of a kind says what arrived. +func IntakeSubject(what, kind string) string { + return "mesh.seat." + IntakeSeat + ".event." + what + "." + kind +} + +// CodeProof is the one proof verb: a code the operator typed, carried by request and reply, never kept. +const CodeProof = "code" + +// ProofSubject is where a channel of a kind asks a proof. +func ProofSubject(verb, kind string) string { + return "mesh.seat." + IntakeSeat + ".proof." + verb + "." + kind +} + +// A Level is how much proof an option's answer needs (novox/hq ADR 0234 §8, the glossary's assurance level). +type Level string + +const ( + // Acknowledge performs only what any granted principal may already do: silencing, details. No proof. + Acknowledge Level = "acknowledge" + // Approve needs one proof: a verified sender (a linked account, linked an hour or more) or a code. + Approve Level = "approve" + // Destroy needs two proofs, one of them a code. + Destroy Level = "destroy" +) + +// Rank orders the levels; an unknown level ranks above every known one, so it is never taken as less. +func (l Level) Rank() int { + switch l { + case Acknowledge: + return 0 + case Approve: + return 1 + case Destroy: + return 2 + } + return 3 +} + +// Operator is the one role an ask may be answered by today. +const Operator = "operator" + +// Option is one answer an ask offers: a label for the button, what it does in plain words, its level. +type Option struct { + ID string `json:"id"` + Label string `json:"label"` + Does string `json:"does"` + Level Level `json:"level"` + // Binds is the digest of exactly what the asker performs when this option is chosen — the verb, the + // machine and every argument — as ActDigest gives it. It travels in the ask, so the router's warrant, + // which names the ask's digest, names it too: a warrant then authorises that act and no other, and an + // asker whose record of the act changed after it asked finds the digests differ and does nothing. + // Required on an option above acknowledge. + Binds string `json:"binds,omitempty"` +} + +// An Act is what an option binds: named fields, each a string — the verb, the machine, the level, and each +// argument under a name of its own ("arg.delivery"). Flat on purpose: its digest is over these names and +// values alone, never over how a language happens to encode a struct. +type Act map[string]string + +// ActDigest is the digest an asker puts in Option.Binds: SHA-256 over the act's canonical encoding (canonical), +// written "sha256:". The same names and values give the same digest in any language, whatever order +// they were set in; a field renamed, added or emptied gives another. +func ActDigest(act Act) (string, error) { + if len(act) == 0 { + return "", fmt.Errorf("the act cannot be digested: it names nothing") + } + keys := make([]string, 0, len(act)) + for k := range act { + if k == "" { + return "", fmt.Errorf("the act cannot be digested: a field has no name") + } + keys = append(keys, k) + } + sort.Strings(keys) + var b strings.Builder + b.WriteString("novox.act.v1\n") + for _, k := range keys { + canonical(&b, k) + canonical(&b, act[k]) + } + sum := sha256.Sum256([]byte(b.String())) + return "sha256:" + hex.EncodeToString(sum[:]), nil +} + +// canonical writes one value as its length in bytes, a colon, the bytes and a newline: no value can be read +// as another's end or start, so two different sequences of values never encode the same. +func canonical(b *strings.Builder, v string) { + b.WriteString(strconv.Itoa(len(v))) + b.WriteByte(':') + b.WriteString(v) + b.WriteByte('\n') +} + +// Digest is the digest of the ask exactly as its asker published it: its id, words, options with what each +// binds, who answers, and its expiry. The router puts it in the warrant (Warrant.AskDigest), and an asker +// acts only on a warrant whose digest is that of the ask it keeps — so a warrant answers one ask, as the +// person was shown it, and nothing published under the same id before or after. +// +// It is over the ask's named fields in a fixed order, each written canonically, and the expiry as UTC +// RFC 3339 to the nanosecond — never over a language's encoding of the struct, so a field added to Ask +// later changes no digest until it is added here, on purpose. +func (a Ask) Digest() string { + var b strings.Builder + b.WriteString("novox.ask.v1\n") + for _, v := range []string{a.ID, a.Headline, a.Explanation, a.Who, + a.Expires.UTC().Format(time.RFC3339Nano), a.OnExpiry, a.About, strconv.FormatBool(a.Urgent), + strconv.Itoa(len(a.Options))} { + canonical(&b, v) + } + for _, o := range a.Options { + for _, v := range []string{o.ID, o.Label, o.Does, string(o.Level), o.Binds} { + canonical(&b, v) + } + } + sum := sha256.Sum256([]byte(b.String())) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +// Ask is a request for a person's word (novox/hq ADR 0259 §4). +type Ask struct { + // ID is the asker's own, unique to it. + ID string `json:"id"` + // Headline names the thing and what is wrong, in a few plain words; Explanation is what happened and + // what it means. Both are held to the plain rule and the content rule by the router. + Headline string `json:"headline"` + Explanation string `json:"explanation"` + Options []Option `json:"options"` + // Who may answer: Operator. + Who string `json:"who"` + // Expires is when the ask ends unanswered; OnExpiry is what the asker then does, in words the person + // is shown ("the delivery stays held"). An ask that authorises never defaults. + Expires time.Time `json:"expires"` + OnExpiry string `json:"on-expiry"` + // About is what the ask is about (a condition's key): a newer ask about it replaces the older. + About string `json:"about,omitempty"` + Urgent bool `json:"urgent,omitempty"` +} + +// The bounds of an ask (novox/hq ADR 0234 §8, ADR 0259 §4). +const ( + MostOptions = 4 + MostOpen = 3 + ApproveLasts = 24 * time.Hour + DestroyLasts = 10 * time.Minute + HeadlineLength = 60 + LabelLength = 24 +) + +var usableID = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$`) + +// UsableID says whether a name can be an ask's or an option's id: a key and a subject token both. +func UsableID(id string) bool { return usableID.MatchString(id) } + +// Highest is the highest level among the ask's options. +func (a Ask) Highest() Level { + high := Acknowledge + for _, o := range a.Options { + if o.Level.Rank() > high.Rank() { + high = o.Level + } + } + return high +} + +// Option is the option of this id, or false. +func (a Ask) Option(id string) (Option, bool) { + for _, o := range a.Options { + if o.ID == id { + return o, true + } + } + return Option{}, false +} + +// Check is what an ask is held to before anything is shown: every refusal, in words its asker can act on. +func (a Ask) Check(now time.Time) error { + var problems []string + say := func(format string, args ...any) { problems = append(problems, fmt.Sprintf(format, args...)) } + if !UsableID(a.ID) { + say("its id %q is not letters, digits, - and _, at most 64", a.ID) + } + if strings.TrimSpace(a.Headline) == "" || len([]rune(a.Headline)) > HeadlineLength { + say("its headline is empty or longer than %d characters", HeadlineLength) + } + if strings.TrimSpace(a.Explanation) == "" { + say("it explains nothing") + } + if a.Who != Operator { + say("it is answered by %q, and only the operator answers today", a.Who) + } + if len(a.Options) == 0 || len(a.Options) > MostOptions { + say("it offers %d options, and an ask offers one to %d", len(a.Options), MostOptions) + } + seen := map[string]bool{} + for _, o := range a.Options { + switch { + case !UsableID(o.ID): + say("an option's id %q is not letters, digits, - and _", o.ID) + case seen[o.ID]: + say("the option %s is offered twice", o.ID) + } + seen[o.ID] = true + if strings.TrimSpace(o.Label) == "" || len([]rune(o.Label)) > LabelLength { + say("the option %s's label is empty or longer than %d characters", o.ID, LabelLength) + } + if strings.TrimSpace(o.Does) == "" { + say("the option %s does not say what it does", o.ID) + } + if o.Level.Rank() > Destroy.Rank() { + say("the option %s has the level %q, which is none of acknowledge, approve, destroy", o.ID, o.Level) + } + if o.Level != Acknowledge && !strings.HasPrefix(o.Binds, "sha256:") { + say("the option %s authorises and does not bind what it performs (its binds is not an ActDigest)", o.ID) + } + } + if !a.Expires.After(now) { + say("it expires before it is asked") + } + switch a.Highest() { + case Approve: + if a.Expires.After(now.Add(ApproveLasts)) { + say("an ask that approves lasts at most %s", ApproveLasts) + } + case Destroy: + if a.Expires.After(now.Add(DestroyLasts)) { + say("an ask that destroys lasts at most %s", DestroyLasts) + } + } + if a.Highest() != Acknowledge && strings.TrimSpace(a.OnExpiry) == "" { + say("it does not say what happens when nobody answers, and an ask that authorises never defaults") + } + if a.About != "" && strings.ContainsAny(a.About, " \n") { + say("what it is about is a key, without spaces") + } + if len(problems) > 0 { + return errors.New("the ask is refused: " + strings.Join(problems, "; ")) + } + return nil +} + +// Outcome is how an ask ended. +type Outcome string + +const ( + OutcomeChosen Outcome = "chosen" // a person chose an option: a warrant + OutcomeExpired Outcome = "expired" // nobody answered in time + OutcomeCancelled Outcome = "cancelled" // its asker took it back + OutcomeReplaced Outcome = "replaced" // a newer ask about the same thing replaced it + OutcomeRefused Outcome = "refused" // it was never shown: Words says why +) + +// Person is who chose, as the router verified them. +type Person struct { + // Who is the role: Operator. + Who string `json:"who"` + // Kind is the channel kind they answered on, Identity their account on that service, Display the name + // the service shows, and Verified how the router knew it was them. + Kind string `json:"kind"` + Identity string `json:"identity"` + Display string `json:"display,omitempty"` + Verified string `json:"verified"` +} + +// Warrant is the router's record that a person chose one option of one ask, or the ask's end without one +// (novox/hq ADR 0259 §6). It carries no secret. +type Warrant struct { + Ask string `json:"ask"` + Asker string `json:"asker"` + About string `json:"about,omitempty"` + Outcome Outcome `json:"outcome"` + // Option, Label and Level are the option chosen; By who chose it, Channel the module it came through, + // Proofs which proofs were present (P1, P2, P3). + Option string `json:"option,omitempty"` + Label string `json:"label,omitempty"` + Level Level `json:"level,omitempty"` + By *Person `json:"by,omitempty"` + Channel string `json:"channel,omitempty"` + Proofs []string `json:"proofs,omitempty"` + At time.Time `json:"at"` + // AskDigest is the digest of the ask as the router took it (Ask.Digest): the warrant answers that ask + // alone, with the options it bound. + AskDigest string `json:"ask-digest,omitempty"` + // Words are why an ask ended without a choice, or what refused it. + Words string `json:"words,omitempty"` +} + +// Says is the warrant in the words an asker records with its act: "the operator, via telegram (user id +// verified), chose Release". +func (w Warrant) Says() string { + if w.Outcome != OutcomeChosen || w.By == nil { + return fmt.Sprintf("no person chose: the ask %s %s", w.Ask, w.Outcome) + } + via := w.By.Kind + if w.By.Verified != "" { + via += " (" + w.By.Verified + ")" + } + return fmt.Sprintf("the %s, via %s, chose %s", w.By.Who, via, w.Label) +} + +// For checks a warrant against the ask its asker made: the same asker and ask, the same ask's digest (so the +// same words, options and binds), a choice, an option the ask offered, at that option's level, before the +// ask expired. An asker acts on nothing else, and then performs only what the option's Binds names. +func (w Warrant) For(asker string, a Ask) (Option, error) { + if w.Asker != asker || w.Ask != a.ID { + return Option{}, fmt.Errorf("the warrant is for %s's ask %s, not %s's %s", w.Asker, w.Ask, asker, a.ID) + } + if w.Outcome != OutcomeChosen || w.By == nil || w.By.Who != a.Who { + return Option{}, fmt.Errorf("the ask %s ended %s; no person chose", a.ID, w.Outcome) + } + if d := a.Digest(); w.AskDigest != d { + return Option{}, fmt.Errorf("the warrant answers an ask whose digest is %q, and the ask %s kept here is %s: "+ + "it was not the ask the person was shown", w.AskDigest, a.ID, d) + } + o, offered := a.Option(w.Option) + if !offered { + return Option{}, fmt.Errorf("the ask %s offered no option %s", a.ID, w.Option) + } + if w.Level != o.Level { + return Option{}, fmt.Errorf("the option %s is %s, and the warrant says %s", o.ID, o.Level, w.Level) + } + // A choice made after the ask expired is no answer to it, whatever the router said. An ask that says no + // expiry, or a warrant that says no time, is no answer either: neither can be shown to be in time (the + // confirmation review of 2026-10-09). + if a.Expires.IsZero() { + return Option{}, fmt.Errorf("the ask %s says no expiry, so no answer to it can be in time", a.ID) + } + if w.At.IsZero() { + return Option{}, fmt.Errorf("the warrant for the ask %s says no time it was given, so it cannot be shown to be in time", a.ID) + } + if w.At.After(a.Expires) { + return Option{}, fmt.Errorf("the warrant was given at %s, after the ask %s expired at %s", + w.At.UTC().Format(time.RFC3339), a.ID, a.Expires.UTC().Format(time.RFC3339)) + } + return o, nil +} + +// Performs checks that the act an asker is about to perform is the one the chosen option bound when it +// asked: the act's digest equals the option's Binds. An acknowledge option that bound nothing passes. +func (o Option) Performs(act Act) error { + if o.Binds == "" && o.Level == Acknowledge { + return nil + } + d, err := ActDigest(act) + if err != nil { + return err + } + if d != o.Binds { + return fmt.Errorf("the option %s bound %s, and the act about to be performed is %s: nothing is done", o.ID, o.Binds, d) + } + return nil +} + +// Button is one answer a channel offers: its label and the router's one-time ticket for it. +type Button struct { + Label string `json:"label"` + Ticket string `json:"ticket"` +} + +// Message is the work a channel takes: shown with buttons (Show), shown again in place (Edit), or said +// (Send). Handle is the router's name for it, the same across a show and its edits; the channel keeps +// which of its own messages that is. The words are the router's, shown as given. +type Message struct { + Handle string `json:"handle"` + Title string `json:"title"` + Body string `json:"body"` + Buttons []Button `json:"buttons,omitempty"` + Urgent bool `json:"urgent,omitempty"` + Silent bool `json:"silent,omitempty"` + // Reply is the Choice or LinkAsked this answers, by its ID: the channel shows it where that was made. + Reply string `json:"reply,omitempty"` + // To is the account linked as the operator on this kind, for a channel that verifies its sender: where + // the channel sends what is not a reply. The router's word, from its list; empty when none is linked. + To string `json:"to,omitempty"` + // Secret says the words carry something shown once (a link's code): the channel shows it and keeps no + // copy of it — no state, no history of its own. + Secret bool `json:"secret,omitempty"` +} + +// Sender is who a channel's service says sent something: the account, the name it shows, and whether the +// service authenticated it. The router alone judges whether that is the operator. +type Sender struct { + Identity string `json:"identity"` + Display string `json:"display,omitempty"` + Authenticated bool `json:"authenticated"` +} + +// Failed is a message a channel could not deliver: its handle, why, and whether trying again could help. +type Failed struct { + Handle string `json:"handle"` + Why string `json:"why"` + Permanent bool `json:"permanent,omitempty"` + At time.Time `json:"at"` +} + +// Standing is what a channel says of itself, at least every five minutes and whenever it changes: +// whether it can send now, why not, and whether its edits notify nobody. +type Standing struct { + Ready bool `json:"ready"` + Why string `json:"why,omitempty"` + EditsSilently bool `json:"edits-silently,omitempty"` + At time.Time `json:"at"` +} + +// What a channel says of its own delivery, on IntakeSubject. +const ( + FailedWhat = "failed" + StandingWhat = "standing" +) + +// Choice is a button chosen on a channel. +type Choice struct { + // ID is the channel's own for this arrival, unique, so the router acts on it once. + ID string `json:"id"` + Ticket string `json:"ticket"` + Handle string `json:"handle,omitempty"` + Sender Sender `json:"sender"` + At time.Time `json:"at"` +} + +// LinkAsked is somebody on a channel asking to be linked as the operator. +type LinkAsked struct { + ID string `json:"id"` + Sender Sender `json:"sender"` + At time.Time `json:"at"` +} + +// Code is a code a person typed on a channel, asked as a proof: never in an event, never kept. +type Code struct { + Sender Sender `json:"sender"` + Code string `json:"code"` + Purpose string `json:"purpose"` +} + +// ProofAnswer is the router's answer to a proof: whether it was taken, and words to say to the person. +type ProofAnswer struct { + Accepted bool `json:"accepted"` + Words string `json:"words"` +} diff --git a/go/asks/asks_test.go b/go/asks/asks_test.go new file mode 100644 index 0000000..999b418 --- /dev/null +++ b/go/asks/asks_test.go @@ -0,0 +1,259 @@ +package asks + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "reflect" + "strings" + "testing" + "time" +) + +var now = time.Date(2026, 10, 8, 14, 0, 0, 0, time.UTC) + +func held() Ask { + return Ask{ID: "d-1", Headline: "openrazer delivery held", Explanation: "Needs you: release it, or stop it.", + Who: Operator, Expires: now.Add(time.Hour), OnExpiry: "the delivery stays held", About: "delivery.openrazer", + Options: []Option{ + {ID: "release", Label: "Release", Does: "the delivery goes on", Level: Approve, Binds: bind("release", "openrazer")}, + {ID: "stop", Label: "Stop", Does: "the delivery ends", Level: Approve, Binds: bind("stop", "openrazer")}, + {ID: "silence", Label: "Silence for a week", Does: "nothing more is said for a week", Level: Acknowledge}, + }} +} + +func bind(verb, what string) string { + d, err := ActDigest(Act{"verb": verb, "delivery": what}) + if err != nil { + panic(err) + } + return d +} + +func TestASoundAskPasses(t *testing.T) { + if err := held().Check(now); err != nil { + t.Fatal(err) + } +} + +func TestAnAskIsRefusedInWordsForEachFault(t *testing.T) { + for want, change := range map[string]func(*Ask){ + "is not letters": func(a *Ask) { a.ID = "a.b" }, + "headline is empty": func(a *Ask) { a.Headline = "" }, + "explains nothing": func(a *Ask) { a.Explanation = " " }, + "only the operator answers": func(a *Ask) { a.Who = "agent" }, + "offers 0 options": func(a *Ask) { a.Options = nil }, + "offers 5 options": func(a *Ask) { a.Options = append(a.Options, a.Options[0], a.Options[1]) }, + "is offered twice": func(a *Ask) { a.Options[1].ID = "release" }, + "does not say what it does": func(a *Ask) { a.Options[0].Does = "" }, + "none of acknowledge": func(a *Ask) { a.Options[0].Level = "admin" }, + "expires before it is asked": func(a *Ask) { a.Expires = now }, + "approves lasts at most": func(a *Ask) { a.Expires = now.Add(25 * time.Hour) }, + "destroys lasts at most": func(a *Ask) { a.Options[0].Level = Destroy }, + "never defaults": func(a *Ask) { a.OnExpiry = "" }, + "is a key, without spaces": func(a *Ask) { a.About = "a b" }, + "does not bind what it": func(a *Ask) { a.Options[0].Binds = "" }, + } { + a := held() + change(&a) + if err := a.Check(now); err == nil || !strings.Contains(err.Error(), want) { + t.Errorf("want a refusal saying %q, got %v", want, err) + } + } + ack := held() + ack.Options = ack.Options[2:] + ack.OnExpiry = "" + ack.Expires = now.Add(7 * 24 * time.Hour) + if err := ack.Check(now); err != nil { + t.Errorf("an ask that only acknowledges needs no expiry words and may last a week: %v", err) + } +} + +func TestAnAskerActsOnlyOnAWarrantForItsOwnAskAndOption(t *testing.T) { + w := Warrant{Ask: "d-1", Asker: "mesh-delivery", Outcome: OutcomeChosen, Option: "release", Label: "Release", + Level: Approve, By: &Person{Who: Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"}, + AskDigest: held().Digest(), At: now.Add(time.Minute)} + if o, err := w.For("mesh-delivery", held()); err != nil || o.ID != "release" { + t.Fatalf("a sound warrant was refused: %v", err) + } + if got := w.Says(); got != "the operator, via telegram (user id verified), chose Release" { + t.Errorf("the warrant says %q", got) + } + for why, change := range map[string]func(*Warrant){ + "another asker": func(w *Warrant) { w.Asker = "mesh-controller" }, + "another ask": func(w *Warrant) { w.Ask = "d-2" }, + "no choice": func(w *Warrant) { w.Outcome = OutcomeExpired }, + "nobody": func(w *Warrant) { w.By = nil }, + "not offered": func(w *Warrant) { w.Option = "delete" }, + "a lower level": func(w *Warrant) { w.Level = Acknowledge }, + "another role": func(w *Warrant) { w.By = &Person{Who: "agent"} }, + "no digest": func(w *Warrant) { w.AskDigest = "" }, + "no time given": func(w *Warrant) { w.At = time.Time{} }, + "after expiry": func(w *Warrant) { w.At = now.Add(2 * time.Hour) }, + "another ask's digest": func(w *Warrant) { + other := held() + other.Options[0].Binds = bind("release", "another") + w.AskDigest = other.Digest() + }, + } { + c := w + by := *w.By + c.By = &by + change(&c) + if _, err := c.For("mesh-delivery", held()); err == nil { + t.Errorf("a warrant with %s was acted on", why) + } + } +} + +func TestTheSubjectsNameTheirPublisher(t *testing.T) { + for got, want := range map[string]string{ + AskSubject("mesh-delivery"): "mesh.seat.operator-channel.accept.ask.mesh-delivery", + DecidedSubject("mesh-delivery"): "mesh.seat.operator-channel.event.decided.mesh-delivery", + ChannelSubject(Show, "telegram"): "mesh.seat.channel.accept.show.telegram", + IntakeSubject(Chosen, "desktop"): "mesh.seat.intake.event.choice.desktop", + ProofSubject(CodeProof, "telegram"): "mesh.seat.intake.proof.code.telegram", + } { + if got != want { + t.Errorf("%s, want %s", got, want) + } + } +} + +// A warrant answers one ask as it was shown, and authorises the act its option bound and no other +// (novox/hq ADR 0259 §6: the warrant is bound to the exact ask and its arguments). +func TestAWarrantAuthorisesOnlyTheActItsOptionBound(t *testing.T) { + a := held() + // The digest survives the trip through the bus: what the router decodes and digests is what was sent. + raw, err := json.Marshal(a) + if err != nil { + t.Fatal(err) + } + var decoded Ask + if err := json.Unmarshal(raw, &decoded); err != nil { + t.Fatal(err) + } + if decoded.Digest() != a.Digest() { + t.Fatalf("the ask's digest changed on the way: %s, %s", decoded.Digest(), a.Digest()) + } + local := a + local.Expires = a.Expires.In(time.FixedZone("CEST", 2*3600)) + if local.Digest() != a.Digest() { + t.Error("the same instant in another zone digests differently") + } + // Changing anything the person was shown, or what an option binds, changes the digest. + for what, change := range map[string]func(*Ask){ + "the headline": func(a *Ask) { a.Headline += "!" }, + "the explanation": func(a *Ask) { a.Explanation += " " }, + "a label": func(a *Ask) { a.Options[0].Label = "Go" }, + "what it binds": func(a *Ask) { a.Options[0].Binds = bind("release", "another") }, + "the expiry": func(a *Ask) { a.Expires = a.Expires.Add(time.Second) }, + "what it is about": func(a *Ask) { a.About = "delivery.other" }, + } { + c := held() + change(&c) + if c.Digest() == a.Digest() { + t.Errorf("changing %s leaves the digest", what) + } + } + o, _ := a.Option("release") + if err := o.Performs(Act{"verb": "release", "delivery": "openrazer"}); err != nil { + t.Errorf("the bound act was refused: %v", err) + } + if err := o.Performs(Act{"verb": "release", "delivery": "another"}); err == nil { + t.Error("an act the option did not bind was let through") + } + ack, _ := a.Option("silence") + if err := ack.Performs(Act{"anything": "x"}); err != nil { + t.Errorf("an acknowledgement that bound nothing: %v", err) + } +} + +// The digests are over named fields in a canonical encoding (the review of 2026-10-09): not over Go's +// encoding of a struct, so a field added to Ask or Option, a JSON tag renamed, or a map built in another +// order changes nothing it should not — and every field that is shown or bound is in it. +func TestDigestsAreCanonicalOverNamedFields(t *testing.T) { + a1, _ := ActDigest(Act{"verb": "release", "arg.delivery": "hq"}) + b := Act{} + b["arg.delivery"] = "hq" + b["verb"] = "release" + a2, _ := ActDigest(b) + if a1 != a2 { + t.Error("the order an act was built in changed its digest") + } + // A value cannot be read as two: "a"+"b=c" is not "a=b"+"c". + x, _ := ActDigest(Act{"a": "b\n1:c"}) + y, _ := ActDigest(Act{"a": "b", "c": ""}) + if x == y { + t.Error("two different acts encode the same") + } + if _, err := ActDigest(Act{}); err == nil { + t.Error("an empty act was digested") + } + // The encoding, by hand: any language reproduces it. + sum := sha256.Sum256([]byte("novox.act.v1\n4:verb\n7:release\n")) + if got, _ := ActDigest(Act{"verb": "release"}); got != "sha256:"+hex.EncodeToString(sum[:]) { + t.Errorf("the act's encoding is not the one written down: %s", got) + } + + // Every exported field of Ask and Option is in the digest: changing any one changes it. + base := held() + mutations := map[string]func(*Ask){ + "ID": func(a *Ask) { a.ID = "other" }, + "Headline": func(a *Ask) { a.Headline += "!" }, + "Explanation": func(a *Ask) { a.Explanation += "!" }, + "Who": func(a *Ask) { a.Who = "someone" }, + "Expires": func(a *Ask) { a.Expires = a.Expires.Add(time.Nanosecond) }, + "OnExpiry": func(a *Ask) { a.OnExpiry += "!" }, + "About": func(a *Ask) { a.About += "!" }, + "Urgent": func(a *Ask) { a.Urgent = !a.Urgent }, + "Options": func(a *Ask) { a.Options = a.Options[:1] }, + "Option.ID": func(a *Ask) { a.Options = append([]Option(nil), a.Options...); a.Options[0].ID = "x" }, + "Option.Label": func(a *Ask) { + a.Options = append([]Option(nil), a.Options...) + a.Options[0].Label += "!" + }, + "Option.Does": func(a *Ask) { a.Options = append([]Option(nil), a.Options...); a.Options[0].Does += "!" }, + "Option.Level": func(a *Ask) { + a.Options = append([]Option(nil), a.Options...) + a.Options[0].Level = Destroy + }, + "Option.Binds": func(a *Ask) { + a.Options = append([]Option(nil), a.Options...) + a.Options[0].Binds = "sha256:00" + }, + } + for name, mutate := range mutations { + c := base + mutate(&c) + if c.Digest() == base.Digest() { + t.Errorf("changing %s left the digest the same", name) + } + } + // A field added to Ask or Option without being added to Digest fails here. + if n := reflect.TypeOf(Ask{}).NumField(); n != 9 { + t.Errorf("Ask has %d fields, and Digest covers 9: add the new one to Digest and to this test", n) + } + if n := reflect.TypeOf(Option{}).NumField(); n != 5 { + t.Errorf("Option has %d fields, and Digest covers 5: add the new one to Digest and to this test", n) + } + // The same instant in another zone is the same ask. + z := base + z.Expires = z.Expires.In(time.FixedZone("x", 3600)) + if z.Digest() != base.Digest() { + t.Error("the expiry's zone changed the digest") + } +} + +// An ask that says no expiry is answered by no warrant (the confirmation review of 2026-10-09): nothing could +// show the answer was in time. +func TestAWarrantForAnAskWithNoExpiryIsRefused(t *testing.T) { + a := held() + a.Expires = time.Time{} + w := Warrant{Ask: "d-1", Asker: "mesh-delivery", Outcome: OutcomeChosen, Option: "release", Label: "Release", + Level: Approve, By: &Person{Who: Operator, Kind: "telegram", Identity: "42"}, AskDigest: a.Digest(), + At: now.Add(time.Minute)} + if _, err := w.For("mesh-delivery", a); err == nil { + t.Error("a warrant for an ask with no expiry was taken") + } +} diff --git a/go/seat.go b/go/seat.go new file mode 100644 index 0000000..aa2f144 --- /dev/null +++ b/go/seat.go @@ -0,0 +1,153 @@ +package stdio + +import ( + "encoding/json" + "strings" + "time" +) + +// Seats that name their caller or their kind (novox/hq ADR 0259 §3): a module submits an accept or says an +// event under its own name or kind, takes a held seat's work, asks a proof and answers one, and reads a +// record a seat keeps for it. The runtime does each as the module, only as far as its membership lists, +// and refuses the rest with the reason. + +// Work is one piece of a seat's work this bundle takes: the subject it was submitted on, its body, and the +// headers the submitter's runtime set (x-source is the submitting module, x-event-id its id). +type Work struct { + Worker string `json:"worker,omitempty"` + Subject string `json:"subject"` + Body json.RawMessage `json:"body"` + Headers map[string]string `json:"headers,omitempty"` +} + +const seatTimeout = 30 * time.Second + +// SeatPublish submits an accept, or says an event, on a seat, de-duplicated by id: the same id twice is +// one message. An empty id is the runtime's to make. +func SeatPublish(subject string, body any, id string) error { + c, err := live() + if err != nil { + return err + } + _, err = c.ask("mesh/seat.publish", map[string]any{"subject": subject, "body": body, "id": id}, seatTimeout) + return err +} + +// SeatProve asks a proof — core request and reply, never kept — and returns its answer. +func SeatProve(subject string, body any) (json.RawMessage, error) { + c, err := live() + if err != nil { + return nil, err + } + return c.ask("mesh/seat.prove", map[string]any{"subject": subject, "body": body}, seatTimeout) +} + +// SeatAnswer answers every proof on a subject (a pattern, `*` one token) with what handle returns. +func SeatAnswer(subject string, handle func(subject string, body json.RawMessage) (any, error)) error { + c, err := live() + if err != nil { + return err + } + c.mu.Lock() + c.provers[subject] = handle + c.mu.Unlock() + _, err = c.ask("mesh/seat.answer", map[string]any{"subject": subject}, seatTimeout) + return err +} + +// SeatTake takes a held seat's work from a worker, handing each piece to handle: taken when it returns +// nil, offered again after an error. Make the work safe to do twice, or skip an x-event-id already done. +func SeatTake(worker string, handle func(Work) error) error { + c, err := live() + if err != nil { + return err + } + c.mu.Lock() + c.takers[worker] = handle + c.mu.Unlock() + _, err = c.ask("mesh/seat.take", map[string]any{"worker": worker}, seatTimeout) + return err +} + +// SeatRecord reads one key of a record a seat keeps for this module, under its own name, or nil. +func SeatRecord(bucket, key string) (*StateEntry, error) { + c, err := live() + if err != nil { + return nil, err + } + raw, err := c.ask("mesh/seat.record", map[string]any{"bucket": bucket, "key": key}, seatTimeout) + if err != nil || len(raw) == 0 || string(raw) == "null" { + return nil, err + } + var e StateEntry + return &e, json.Unmarshal(raw, &e) +} + +// KindHeld is one kind of a kinded bench and who holds it, with what it promises (novox/hq ADR 0259 §5). +type KindHeld struct { + Seat string `json:"seat"` + Kind string `json:"kind"` + Module string `json:"module"` + Node string `json:"node"` + Capabilities []string `json:"capabilities,omitempty"` +} + +// Promises says whether this kind's claim declares a capability. +func (k KindHeld) Promises(capability string) bool { + for _, c := range k.Capabilities { + if c == capability { + return true + } + } + return false +} + +// SeatKinds is every kind of the kinded benches this module uses or watches, as the controller issued +// them in its membership: the one account of which channel is which, never a channel's own word. +func SeatKinds() ([]KindHeld, error) { + c, err := live() + if err != nil { + return nil, err + } + raw, err := c.ask("mesh/seat.kinds", map[string]any{}, seatTimeout) + if err != nil { + return nil, err + } + var out []KindHeld + return out, json.Unmarshal(raw, &out) +} + +// Create writes a key only when it has none, and returns the revision; a key that has one is refused. +func (s KeptState) Create(key string, value any) (uint64, error) { + return s.revisionOf(s.ask("create", map[string]any{"key": key, "value": value})) +} + +// Update writes a key only when it is still at revision, and returns the new one; a key changed +// meanwhile is refused, so of two writers that read one value only the first writes. +func (s KeptState) Update(key string, value any, revision uint64) (uint64, error) { + return s.revisionOf(s.ask("update", map[string]any{"key": key, "value": value, "revision": revision})) +} + +func (s KeptState) revisionOf(raw json.RawMessage, err error) (uint64, error) { + if err != nil { + return 0, err + } + var put struct { + Revision uint64 `json:"revision"` + } + return put.Revision, json.Unmarshal(raw, &put) +} + +// subjectMatches is a subject against a pattern in the bus's wildcards: `*` one token, `>` the rest. +func subjectMatches(pattern, subject string) bool { + p, s := strings.Split(pattern, "."), strings.Split(subject, ".") + for i, part := range p { + if part == ">" { + return len(s) > i + } + if i >= len(s) || (part != "*" && part != s[i]) { + return false + } + } + return len(p) == len(s) +} diff --git a/go/seat_test.go b/go/seat_test.go new file mode 100644 index 0000000..933b549 --- /dev/null +++ b/go/seat_test.go @@ -0,0 +1,98 @@ +package stdio + +import ( + "bufio" + "encoding/json" + "errors" + "io" + "testing" + "time" +) + +// novox/hq ADR 0259 §3: a bundle takes a seat's work and answers proofs through the runtime; a piece of +// work it fails is answered as an error, so the runtime offers it again, and a proof's error is said. +func TestABundleTakesWorkAndAnswersProofsThroughTheRuntime(t *testing.T) { + t.Setenv(ServedAs, "router") + inR, inW := io.Pipe() + outR, outW := io.Pipe() + taken := make(chan string, 4) + go func() { + _ = ServeOn("", []Tool{{Name: "start", Run: func(map[string]any) (any, error) { + if err := SeatTake("SEAT_OPERATOR_CHANNEL_worker", func(w Work) error { + taken <- w.Subject + if string(w.Body) == `{"fail":true}` { + return errors.New("not recorded") + } + return nil + }); err != nil { + return nil, err + } + if err := SeatAnswer("mesh.seat.intake.proof.code.*", func(subject string, body json.RawMessage) (any, error) { + if subject != "mesh.seat.intake.proof.code.telegram" { + return nil, errors.New("no such kind") + } + return map[string]any{"accepted": true}, nil + }); err != nil { + return nil, err + } + return map[string]any{}, nil + }}}, inR, outW) + }() + lines := bufio.NewScanner(outR) + next := func() map[string]any { + done := make(chan map[string]any, 1) + go func() { + if lines.Scan() { + var m map[string]any + _ = json.Unmarshal(lines.Bytes(), &m) + done <- m + } + }() + select { + case m := <-done: + return m + case <-time.After(3 * time.Second): + t.Fatal("the bundle said nothing") + return nil + } + } + send := func(m map[string]any) { b, _ := json.Marshal(m); _, _ = inW.Write(append(b, '\n')) } + + send(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": map[string]any{"name": "start"}}) + asked := next() + if asked["method"] != "mesh/seat.take" || asked["params"].(map[string]any)["worker"] != "SEAT_OPERATOR_CHANNEL_worker" { + t.Fatalf("the take was not asked of the runtime: %v", asked) + } + send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{}}) + asked = next() + if asked["method"] != "mesh/seat.answer" { + t.Fatalf("the answer was not asked of the runtime: %v", asked) + } + send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{}}) + next() // the tool's answer + + send(map[string]any{"jsonrpc": "2.0", "id": 2, "method": "mesh/work", "params": map[string]any{"work": map[string]any{ + "worker": "SEAT_OPERATOR_CHANNEL_worker", "subject": "mesh.seat.operator-channel.accept.ask.x", "body": map[string]any{"fail": true}}}}) + if got := next(); got["error"] == nil { + t.Errorf("work the handler failed was answered as taken: %v", got) + } + send(map[string]any{"jsonrpc": "2.0", "id": 3, "method": "mesh/work", "params": map[string]any{"work": map[string]any{ + "worker": "SEAT_OPERATOR_CHANNEL_worker", "subject": "mesh.seat.operator-channel.accept.ask.x", "body": map[string]any{}}}}) + if got := next(); got["error"] != nil { + t.Errorf("work the handler took was answered as an error: %v", got) + } + send(map[string]any{"jsonrpc": "2.0", "id": 4, "method": "mesh/proof", "params": map[string]any{ + "subject": "mesh.seat.intake.proof.code.telegram", "body": map[string]any{"code": "1"}}}) + if got := next(); got["result"].(map[string]any)["accepted"] != true { + t.Errorf("the proof was answered %v", got) + } + send(map[string]any{"jsonrpc": "2.0", "id": 5, "method": "mesh/proof", "params": map[string]any{ + "subject": "mesh.seat.intake.proof.code.desktop", "body": map[string]any{}}}) + if got := next(); got["error"] == nil { + t.Errorf("a proof the handler refused was answered as accepted: %v", got) + } + if len(taken) != 2 { + t.Errorf("the handler was handed %d pieces of work, want 2", len(taken)) + } + _ = inW.Close() +} diff --git a/go/stdio.go b/go/stdio.go index 310d76a..3582f14 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -48,6 +48,7 @@ type message struct { Params json.RawMessage `json:"params,omitempty"` Result json.RawMessage `json:"result,omitempty"` Error *struct { + Code int `json:"code"` Message string `json:"message"` } `json:"error,omitempty"` } @@ -64,6 +65,9 @@ type channel struct { subscribed int // watches are this bundle's state watches by the id each was asked with (ADR 0201). watches map[string]func(StateChange) error + // takers and provers are this bundle's seat work and proof handlers, by worker and subject (ADR 0259). + takers map[string]func(Work) error + provers map[string]func(subject string, body json.RawMessage) (any, error) } type listener struct { @@ -111,7 +115,7 @@ func (c *channel) askAs(id, method string, params any, timeout time.Duration) (j select { case m := <-answer: if m.Error != nil { - return nil, &refused{m.Error.Message} + return nil, &refused{message: m.Error.Message, code: m.Error.Code} } return m.Result, nil case <-time.After(timeout): @@ -123,10 +127,23 @@ func (c *channel) askAs(id, method string, params any, timeout time.Duration) (j } // refused is the runtime's error answer, as opposed to no answer: what it refused, it did not do. -type refused struct{ message string } +type refused struct { + message string + code int +} func (r *refused) Error() string { return r.message } +// Is makes a compare-and-set the runtime refused because another writer was first ErrChanged. +func (r *refused) Is(target error) bool { return target == ErrChanged && r.code == CodeChanged } + +// ErrChanged is a compare-and-set that lost: the key was made or changed by another writer first (novox/hq +// ADR 0259). Told apart by the runtime's error code, never by its words. +var ErrChanged = errors.New("the key was written by another writer first") + +// CodeChanged is the error code the runtime answers a lost compare-and-set with. +const CodeChanged = -32010 + var ( current *channel currentMu sync.Mutex @@ -150,7 +167,8 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { for _, t := range tools { byName[t.Name] = t } - c := &channel{enc: json.NewEncoder(out), waiting: map[string]chan message{}, watches: map[string]func(StateChange) error{}} + c := &channel{enc: json.NewEncoder(out), waiting: map[string]chan message{}, watches: map[string]func(StateChange) error{}, + takers: map[string]func(Work) error{}, provers: map[string]func(string, json.RawMessage) (any, error){}} currentMu.Lock() current = c currentMu.Unlock() @@ -276,6 +294,63 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { } c.say(map[string]any{"id": id, "result": map[string]any{}}) }(id, *p.Change) + case "mesh/work": + // One piece of a seat's work this bundle takes (novox/hq ADR 0259): taken by an answer that is + // not an error, offered again after one. Beside the loop, like a tool call. + var p struct { + Work *Work `json:"work"` + } + if json.Unmarshal(m.Params, &p) != nil || p.Work == nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/work carries no work"}}) + continue + } + c.mu.Lock() + handle := c.takers[p.Work.Worker] + if handle == nil && len(c.takers) == 1 { + for _, h := range c.takers { + handle = h + } + } + c.mu.Unlock() + go func(id any, w Work) { + if handle == nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": "nothing here takes " + w.Subject}}) + return + } + if err := handle(w); err != nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) + return + } + c.say(map[string]any{"id": id, "result": map[string]any{}}) + }(id, *p.Work) + case "mesh/proof": + // One proof this bundle answers (novox/hq ADR 0259): what the handler returns is the answer; an + // error is answered in words, never as silence. + var p struct { + Subject string `json:"subject"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &p) + c.mu.Lock() + var handle func(string, json.RawMessage) (any, error) + for pattern, h := range c.provers { + if subjectMatches(pattern, p.Subject) { + handle = h + } + } + c.mu.Unlock() + go func(id any, subject string, body json.RawMessage) { + if handle == nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": "nothing here answers " + subject}}) + return + } + out, err := handle(subject, body) + if err != nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) + return + } + c.say(map[string]any{"id": id, "result": out}) + }(id, p.Subject, p.Body) default: if id != nil { c.say(map[string]any{"id": id, "error": map[string]any{"code": -32601, "message": "no " + m.Method}})