From 0292772f58acdc34b60bba414923b7d8752f947e Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 8 Oct 2026 16:34:16 +0200 Subject: [PATCH] Give modules the seat traffic and the contract to ask a person and answer on a channel (hq ADR 0259) --- README.md | 10 ++ go/asks/asks.go | 349 +++++++++++++++++++++++++++++++++++++++++++ go/asks/asks_test.go | 99 ++++++++++++ go/seat.go | 119 +++++++++++++++ go/seat_test.go | 98 ++++++++++++ go/stdio.go | 63 +++++++- 6 files changed, 737 insertions(+), 1 deletion(-) create mode 100644 go/asks/asks.go create mode 100644 go/asks/asks_test.go create mode 100644 go/seat.go create mode 100644 go/seat_test.go diff --git a/README.md b/README.md index 01d1d3e..22d54b8 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,16 @@ 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. + ### `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/asks/asks.go b/go/asks/asks.go new file mode 100644 index 0000000..90be5cf --- /dev/null +++ b/go/asks/asks.go @@ -0,0 +1,349 @@ +// 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 ( + "errors" + "fmt" + "regexp" + "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"` +} + +// 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 !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"` + // 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, a choice, an option the ask +// offered, at that option's level. An asker acts on nothing else. +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) + } + 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) + } + return o, 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"` +} + +// 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"` +} + +// 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..9c6ff89 --- /dev/null +++ b/go/asks/asks_test.go @@ -0,0 +1,99 @@ +package asks + +import ( + "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}, + {ID: "stop", Label: "Stop", Does: "the delivery ends", Level: Approve}, + {ID: "silence", Label: "Silence for a week", Does: "nothing more is said for a week", Level: Acknowledge}, + }} +} + +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" }, + } { + 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"}} + 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"} }, + } { + 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) + } + } +} diff --git a/go/seat.go b/go/seat.go new file mode 100644 index 0000000..7d44806 --- /dev/null +++ b/go/seat.go @@ -0,0 +1,119 @@ +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) +} + +// 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..d0cc09b 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -64,6 +64,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 { @@ -150,7 +153,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 +280,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}})