package broker import "sort" // What a module may say and take on the seats it holds, uses and watches, beyond tools (novox/hq ADR // 0259 §3, to-be 46 §10). // // Three rules are added to the ones a seat always had, each a subject whose last token names who may // publish it, granted to that publisher alone — the way a node seat's event about a machine carries the // machine (ADR 0219): // // - **A verb named by its caller** (`by-caller`). A user of the seat submits that accept, and hears that // event, under its own module's name and no other: `accept.ask.`, `event.decided.`. The // holder takes every caller's accept and says the event to any caller. So an ask's asker is a fact the // server enforces, and a warrant reaches only the asker it is for. // - **A kinded bench** (ADR 0234 §2). A holder claims one kind and takes its own kind's accepts, says its // own kind's events and asks its own kind's proofs, and nothing of another kind; a user submits to any // kind. Each kind has its own worker on the seat's queue, so a holder that is away keeps its work and // holds up no other kind. // - **A proof** (`proofs`): core request and reply on `mesh.seat..proof..`. No stream's // subjects cover it, so what travels there — a code typed by the operator — is never persisted. A kinded // holder asks with its own kind; the modules that watch the seat answer. // // And one read: **a holder's records**, a bucket the seat names, read by each user under its own name only // (`$KV...>`), so an asker reads the state of its own asks and no other asker's. // Worker is one durable consumer a holder pulls a seat's work from. type Worker struct { Stream string Consumer string Filter string } // SeatTraffic is one module's seat traffic beyond tools. Publish and Subscribe are subject patterns, in the // server's wildcards; the runtime that carries the module checks a bundle's request against them, since // the runtime's own principal holds the union of every module it carries. type SeatTraffic struct { // Publish is what it submits (accepts of seats it uses), says (events of seats it holds) and asks // (proofs of seats it holds a kind of). Publish []string `json:"publish,omitempty"` // Subscribe is what it hears: events of seats it uses that are named by caller, events of seats it // watches, and accepts of seats it holds. Subscribe []string `json:"subscribe,omitempty"` // Answers is the proof subjects it answers, as a watcher of a kinded seat. Answers []string `json:"answers,omitempty"` // Workers are the work queues it takes from, as a holder. Workers []Worker `json:"workers,omitempty"` // Records is the direct-get subjects of the records it reads under its own name. Records []string `json:"records,omitempty"` // Kinds are the holders of every kinded bench it uses or watches, with what each promises: the one // account of which channel is which, and what it can carry, that the router judges an answer by. The // controller's, from the claims, never a channel's word (novox/hq ADR 0259 §5). Kinds []KindHeld `json:"kinds,omitempty"` } // KindHeld is one kind of a kinded bench and who holds it. type KindHeld struct { Seat string `json:"seat"` Kind string `json:"kind"` Module string `json:"module"` Node string `json:"node"` Capabilities []string `json:"capabilities,omitempty"` } // KindedBenches are the seats that may be kinded (ADR 0234 §2): making another is a decision, recorded. var KindedBenches = map[string]bool{"channel": true, "intake": true} // WorkerName is the worker a seat's holders pull from: one for the seat, or one per kind on a kinded bench. func WorkerName(seat, kind string) string { if kind == "" { return "SEAT_" + upperSnake(seat) + "_worker" } return "SEAT_" + upperSnake(seat) + "_" + upperSnake(kind) + "_worker" } func namesVerb(list []string, s string) bool { for _, x := range list { if x == s { return true } } return false } // SeatTrafficOf derives one module's seat traffic from the seats it holds, uses and watches. Only seats // carrying one of the rules above are read: every other seat is composed as it always was. func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic { var t SeatTraffic for _, s := range holds { if !s.isNewTraffic() { continue } kind := "" if s.Kinded { kind = s.Kind if kind == "" || !safeSubject.MatchString(kind) { // A kinded claim without a usable kind is refused at registration; here it is granted // nothing, which is the same answer at the last place it could be asked. continue } } if len(s.Accepts) > 0 { stream := seatStreamName(s.Name) filter := "mesh.seat." + s.Name + ".accept.>" if kind != "" { filter = "mesh.seat." + s.Name + ".accept.*." + kind } t.Workers = append(t.Workers, Worker{Stream: stream, Consumer: WorkerName(s.Name, kind), Filter: filter}) } for _, a := range s.Accepts { switch { case kind != "": t.Subscribe = append(t.Subscribe, seatSubject(s, "accept", a+"."+kind)) case namesVerb(s.ByCaller, a): t.Subscribe = append(t.Subscribe, seatSubject(s, "accept", a+".*")) } } for _, e := range s.Emits { switch { case kind != "": t.Publish = append(t.Publish, seatSubject(s, "event", e+"."+kind)) case namesVerb(s.ByCaller, e): t.Publish = append(t.Publish, seatSubject(s, "event", e+".*")) } } if kind != "" { for _, v := range s.Proofs { t.Publish = append(t.Publish, seatSubject(s, "proof", v+"."+kind)) } } } for _, s := range uses { if !s.isNewTraffic() { continue } for _, a := range s.Accepts { switch { case namesVerb(s.ByCaller, a): t.Publish = append(t.Publish, seatSubject(s, "accept", a+"."+module)) case s.Kinded && module == s.DeclaredBy: // Work for a kind is put on its queue by the bench's own router, and by no other user. t.Publish = append(t.Publish, seatSubject(s, "accept", a+".*")) } } for _, e := range s.Emits { if namesVerb(s.ByCaller, e) { t.Subscribe = append(t.Subscribe, seatSubject(s, "event", e+"."+module)) } } for _, b := range s.Records { if !safeSubject.MatchString(b) { continue } t.Records = append(t.Records, "$JS.API.DIRECT.GET.KV_"+b+".$KV."+b+"."+module+".>") } } for _, w := range watches { if w.Kinded { for _, e := range w.Emits { t.Subscribe = append(t.Subscribe, seatSubject(w, "event", e+".*")) } if module == w.DeclaredBy { // A code is answered by the bench's own router, and by no other watcher. for _, v := range w.Proofs { t.Answers = append(t.Answers, seatSubject(w, "proof", v+".*")) } } } } t.Publish = unique(t.Publish) t.Subscribe = unique(t.Subscribe) t.Answers = unique(t.Answers) t.Records = unique(t.Records) sort.Slice(t.Workers, func(i, j int) bool { return t.Workers[i].Consumer < t.Workers[j].Consumer }) return t } // grants is the bus permissions seat traffic needs: the subjects themselves, and the JetStream API a // worker is pulled and acknowledged through and a record is read through. func (t SeatTraffic) grants() (pub, sub []string) { pub = append(pub, t.Publish...) sub = append(sub, t.Subscribe...) sub = append(sub, t.Answers...) for _, w := range t.Workers { pub = append(pub, "$JS.API.CONSUMER.INFO."+w.Stream+"."+w.Consumer, "$JS.API.CONSUMER.MSG.NEXT."+w.Stream+"."+w.Consumer, "$JS.ACK."+w.Stream+"."+w.Consumer+".>") } pub = append(pub, t.Records...) return pub, sub } // isNewTraffic says whether a seat carries any of the rules above, so a seat that carries none is // composed exactly as before them. func (s Seat) isNewTraffic() bool { return s.Kinded || len(s.ByCaller) > 0 || len(s.Proofs) > 0 || len(s.Records) > 0 } // plainVerbs is a seat's accepts or emits with those the rules above compose taken out: a verb named by // its caller and every verb of a kinded bench are composed by SeatTrafficOf and nowhere else. func plainVerbs(s Seat, verbs []string) []string { if s.Kinded { return nil } var out []string for _, v := range verbs { if !namesVerb(s.ByCaller, v) { out = append(out, v) } } return out } // SeatTrafficObjects is the work queues and workers the seat traffic of every composed user implies // (novox/hq ADR 0259 §3): a queue for each seat a holder takes work from, and each holder's worker on it — // one per kind on a kinded bench, filtered to that kind, so the kinds never take each other's work. Only // seats carrying the rules above; the mesh's own seats' queues are RaiseSeats'. func SeatTrafficObjects(users []Principal) ([]Stream, []Consumer) { streams := map[string]Stream{} consumers := map[string]Consumer{} add := func(module string, holds []Seat) { for _, w := range SeatTrafficOf(module, holds, nil, nil).Workers { seat := "" for _, s := range holds { if seatStreamName(s.Name) == w.Stream { seat = s.Name } } streams[w.Stream] = Stream{ Name: w.Stream, Subjects: []string{"mesh.seat." + seat + ".accept.>"}, Retention: RetentionWorkQueue, MaxAge: 7 * 24 * 60 * 60, Why: "work submitted to the " + seat + " seat; its holders take it, each kind its own, and it queues while nobody does", } consumers[w.Consumer] = Consumer{ Name: w.Consumer, Stream: w.Stream, Filters: []string{w.Filter}, AckWaitSeconds: 60, // No bound on redelivery: a channel away for a day keeps its work, offered again later and // later by its holder's runtime (novox/hq ADR 0259; the correctness review of 2026-10-08). MaxDeliver: 0, Why: module + " holds " + seat + "; it pulls one ask at a time and acknowledges once it has " + "recorded it, so a crash redelivers rather than loses", } } } for _, p := range users { switch p.Kind { case KindModule: add(p.Module, p.Holds) case KindNodeTools: for _, d := range p.Carries { add(d.Module, d.Holds) } } } var ss []Stream for _, s := range streams { ss = append(ss, s) } sort.Slice(ss, func(i, j int) bool { return ss[i].Name < ss[j].Name }) var cs []Consumer for _, c := range consumers { cs = append(cs, c) } sort.Slice(cs, func(i, j int) bool { return cs[i].Name < cs[j].Name }) return ss, cs } // trustedTraffic is why a module's seat traffic is the trusted holder's (novox/hq ADR 0259 §8), or "": it // says a seat's event to one caller each (a warrant), or holds a kind of a kinded bench that proves its sender. func trustedTraffic(d Declared) string { for _, s := range d.Holds { for _, e := range s.Emits { if namesVerb(s.ByCaller, e) { return "it says " + s.Name + "'s " + e + " to one caller each" } } if s.Kinded && namesVerb(s.Capabilities, "verified-sender") { return "it holds " + s.Name + " of kind " + s.Kind + ", which proves its sender" } } return "" } // TrafficQueues is the work queue of every seat naming its caller or its kind that accepts work, held or not // (novox/hq ADR 0259 §3): what is submitted before a holder is assigned waits for it. func TrafficQueues(seats []Seat) []Stream { var out []Stream seen := map[string]bool{} for _, s := range seats { if len(s.Accepts) == 0 || !s.isNewTraffic() || seen[s.Name] { continue } seen[s.Name] = true out = append(out, Stream{Name: seatStreamName(s.Name), Subjects: []string{"mesh.seat." + s.Name + ".accept.>"}, Retention: RetentionWorkQueue, MaxAge: 7 * 24 * 60 * 60, Why: "work submitted to the " + s.Name + " seat; its holders take it, each kind its own, and it queues while nobody does"}) } sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) return out }