Let a module's code announce the states it keeps, in Go and TypeScript (issue 1790, S10, ADR 0495 §2 §5)

process.Conn.Keeps, stdio.Keeps and the TypeScript keeps publish the announcement on
the module's own subject; afterwards a state the code did not announce is refused at
the call, so announcement and code cannot drift. The package goes to 0.1.15 for keeps.

go/keeps.go

@@ -0,0 +1,80 @@
+package stdio
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+
+	"git.novox.be/novox/mesh-sdk/go/announce"
+)
+
+// What this bundle said it keeps (novox/hq ADR 0495 clause 2; issue 1790).
+var (
+	keepsMu   sync.Mutex
+	kept      map[string]bool // nil until Keeps: no state is refused then
+	announced []byte          // the announcement, published once the bundle serves
+)
+
+// Keeps says on the bus, as this bundle's module, every state it keeps and which of them others may read (novox/hq
+// ADR 0495 clause 2): what a reader is granted is what the owner announced (clause 3). Call it before Serve; the
+// announcement is published through the runtime once the bundle serves, or at once when it already does. From then
+// on a state of the module's own that it did not announce is refused at the call, naming it, so code and announcement
+// cannot drift (clause 5). It refuses a state no bucket could be named by and a state given twice.
+func Keeps(states ...announce.State) error {
+	module := os.Getenv("MESH_MODULE")
+	if module == "" {
+		module = os.Getenv(ServedAs)
+	}
+	data, err := announce.Encode(announce.States{Module: module, Node: os.Getenv("MESH_NODE"), States: states})
+	if err != nil {
+		return err
+	}
+	names := map[string]bool{}
+	for _, s := range states {
+		names[s.Name] = true
+	}
+	keepsMu.Lock()
+	kept, announced = names, data
+	keepsMu.Unlock()
+	currentMu.Lock()
+	serving := current != nil
+	currentMu.Unlock()
+	if serving {
+		return announce1()
+	}
+	return nil
+}
+
+// announce1 publishes the announcement Keeps made, if any.
+func announce1() error {
+	keepsMu.Lock()
+	data := announced
+	keepsMu.Unlock()
+	if data == nil {
+		return nil
+	}
+	return Emit(announce.Event, json.RawMessage(data))
+}
+
+// refusedUnannounced is the error of a call on a state of the module's own that Keeps did not name, or nil. A read of
+// another module's state (`<module>.<name>`) is the grant's business.
+func refusedUnannounced(name string) error {
+	if strings.Contains(name, ".") {
+		return nil
+	}
+	keepsMu.Lock()
+	defer keepsMu.Unlock()
+	if kept != nil && !kept[name] {
+		return fmt.Errorf("the state %q is not among the states this module announced it keeps (stdio.Keeps)", name)
+	}
+	return nil
+}
+
+// forgetKeeps is for a test: nothing announced.
+func forgetKeeps() {
+	keepsMu.Lock()
+	kept, announced = nil, nil
+	keepsMu.Unlock()
+}

go/keeps_test.go

@@ -0,0 +1,63 @@
+package stdio
+
+import (
+	"bufio"
+	"encoding/json"
+	"io"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/announce"
+)
+
+// A Go bundle says what it keeps from its code (novox/hq ADR 0495 clauses 1 and 2): Keeps, called before Serve,
+// announces through the runtime once the bundle serves, as the module the runtime names; and a state of its own that it
+// did not announce is refused at the call, naming it, so code and announcement cannot drift.
+func TestABundleAnnouncesWhatItKeepsAndBindsNothingElse(t *testing.T) {
+	t.Setenv(ServedAs, "chime")
+	t.Setenv("MESH_MODULE", "chime")
+	t.Setenv("MESH_NODE", "anchor")
+	defer forgetKeeps()
+	// Not yet serving: an earlier test's channel is gone with its test.
+	currentMu.Lock()
+	current = nil
+	currentMu.Unlock()
+	if err := Keeps(announce.State{Name: "rings", Readable: true}, announce.State{Name: "own"}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := State("other").Get("k"); err == nil || !strings.Contains(err.Error(), `"other"`) {
+		t.Fatalf("a state not announced was bound: %v", err)
+	}
+	inR, inW := io.Pipe()
+	outR, outW := io.Pipe()
+	go func() { _ = ServeOn("", nil, inR, outW) }()
+	defer inW.Close()
+	lines := bufio.NewScanner(outR)
+	got := make(chan map[string]any, 1)
+	go func() {
+		if lines.Scan() {
+			var m map[string]any
+			_ = json.Unmarshal(lines.Bytes(), &m)
+			got <- m
+		}
+	}()
+	var asked map[string]any
+	select {
+	case asked = <-got:
+	case <-time.After(3 * time.Second):
+		t.Fatal("the bundle announced nothing")
+	}
+	p, _ := asked["params"].(map[string]any)
+	if asked["method"] != "mesh/publish" || p["key"] != announce.Event {
+		t.Fatalf("the announcement was not published through the runtime: %v", asked)
+	}
+	raw, _ := json.Marshal(p["body"])
+	s, err := announce.Decode(raw)
+	if err != nil || s.Module != "chime" || s.Node != "anchor" || !s.Readable("rings") || !s.Keeps("own") || s.Readable("own") {
+		t.Fatalf("announced %s (%v)", raw, err)
+	}
+	if err := Keeps(announce.State{Name: "bad_name"}); err == nil {
+		t.Fatal("an announcement no bucket could be named by was accepted")
+	}
+}

go/process/conn.go

@@ -24,6 +24,12 @@ type Conn struct {
 	mu      sync.Mutex
 	next    uint64
 	pending map[string]map[uint64]context.CancelCauseFunc
+
+	// What Keeps announced (keeps.go): the states' names (nil before Keeps), the announcement, and whether a
+	// reconnect announces it again.
+	kept        map[string]bool
+	announced   []byte
+	reannounces bool
 }
 
 // errPermissionRefused marks a context cancelled because the bus refused this account's publish of the

go/process/fixture_integration_test.go

@@ -38,6 +38,7 @@ var sendGrants = []string{
 	"$KV.mesh-mail_sends.>",
 	"mesh.mod.mesh-mail.event.mail.send.failed",
 	"mesh.mod.mesh-mail.event.mail.send.sent",
+	"mesh.mod.mesh-mail.event.states",
 	"mesh.seat.operator-channel.call.tell.*.novox~mesh-mail~send",
 	"mesh.seat.operator-channel.call.tell.novox~mesh-mail~send",
 	"mesh.seat.operator-channel.tool.tell",

go/process/keeps.go

@@ -0,0 +1,71 @@
+package process
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/nats-io/nats.go"
+
+	"git.novox.be/novox/mesh-sdk/go/announce"
+)
+
+// Keeps says on the bus, as this process's module, every state it keeps and which of them others may read (novox/hq
+// ADR 0495 clause 2; issue 1790), on announce.Subject; what a reader is granted is what the owner announced (clause 3).
+// It publishes at once and again after every reconnect, so the controller hears it whenever it missed it. From then on
+// a state of the module that it did not announce is refused at the call, naming it, so code and announcement cannot
+// drift (clause 5). It refuses a state no bucket could be named by and a state given twice.
+func (c *Conn) Keeps(states ...announce.State) error {
+	if err := c.remember(states); err != nil {
+		return err
+	}
+	c.mu.Lock()
+	first := !c.reannounces
+	c.reannounces = true
+	c.mu.Unlock()
+	if first {
+		before := c.nc.Opts.ReconnectedCB
+		c.nc.SetReconnectHandler(func(nc *nats.Conn) {
+			if before != nil {
+				before(nc)
+			}
+			_ = c.publishKeeps()
+		})
+	}
+	return c.publishKeeps()
+}
+
+// remember makes the announcement of states and binds only them from now on.
+func (c *Conn) remember(states []announce.State) error {
+	data, err := announce.Encode(announce.States{Module: c.account.Module, Node: c.account.Node, States: states})
+	if err != nil {
+		return err
+	}
+	names := map[string]bool{}
+	for _, s := range states {
+		names[s.Name] = true
+	}
+	c.mu.Lock()
+	c.kept, c.announced = names, data
+	c.mu.Unlock()
+	return nil
+}
+
+func (c *Conn) publishKeeps() error {
+	c.mu.Lock()
+	data := c.announced
+	c.mu.Unlock()
+	if err := c.Emit(announce.Event, json.RawMessage(data)); err != nil {
+		return fmt.Errorf("announcing what %s keeps: %w", c.account.Module, err)
+	}
+	return nil
+}
+
+// refusedUnannounced is the error of a bind of a state Keeps did not name, or nil (also before Keeps).
+func (c *Conn) refusedUnannounced(name string) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.kept != nil && !c.kept[name] {
+		return fmt.Errorf("the state %q is not among the states %s announced it keeps (Conn.Keeps)", name, c.account.Module)
+	}
+	return nil
+}

go/process/keeps_integration_test.go

@@ -0,0 +1,45 @@
+//go:build integration
+
+package process
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/announce"
+)
+
+// 1790/S10: on a real bus, a process's Keeps lands on its module's own announcement subject, in the events stream, as
+// the module's announcement; a state it did not announce is then refused at the call.
+func TestAProcessAnnouncesWhatItKeepsOnTheBus(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	_, admin := adminJS(t, url)
+	stream, err := admin.CreateStream(ctxFor(t), jetstream.StreamConfig{Name: "MAIL_STATES", Subjects: []string{announce.Subject("mesh-mail")}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	if err := conn.Keeps(announce.State{Name: "accounts", Readable: true}, announce.State{Name: "drafts"}); err != nil {
+		t.Fatal(err)
+	}
+	msg, err := stream.GetMsg(ctxFor(t), 1)
+	if err != nil {
+		t.Fatalf("no announcement in the stream: %v", err)
+	}
+	if msg.Subject != "mesh.mod.mesh-mail.event.states" || msg.Header.Get("x-source") != "mesh-mail" {
+		t.Fatalf("announced on %s with %v", msg.Subject, msg.Header)
+	}
+	got, err := announce.Decode(msg.Data)
+	if err != nil || !got.Readable("accounts") || !got.Keeps("drafts") || got.Readable("drafts") || got.Node != "novox" {
+		t.Fatalf("announced %s (%v)", msg.Data, err)
+	}
+	if _, err := conn.State("outbox").Get("k"); err == nil || !strings.Contains(err.Error(), `"outbox"`) {
+		t.Fatalf("an unannounced state was bound: %v", err)
+	}
+}

go/process/keeps_test.go

@@ -0,0 +1,28 @@
+package process
+
+import (
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/announce"
+)
+
+// After Keeps a process binds only the states it announced (novox/hq ADR 0495 clause 5): any other of its module's
+// states is refused at the call, naming it, before the bus is asked; before Keeps nothing is refused.
+func TestAProcessBindsOnlyTheStatesItAnnounced(t *testing.T) {
+	c := &Conn{account: Account{Module: "mesh-mail", Node: "novox"}}
+	if err := c.remember([]announce.State{{Name: "accounts", Readable: true}}); err != nil {
+		t.Fatal(err)
+	}
+	_, err := c.State("drafts").Get("k")
+	if err == nil || !strings.Contains(err.Error(), `"drafts"`) || !strings.Contains(err.Error(), "Keeps") {
+		t.Fatalf("an unannounced state was bound: %v", err)
+	}
+	if err := c.remember([]announce.State{{Name: "a_b"}}); err == nil {
+		t.Fatal("an announcement no bucket could be named by was accepted")
+	}
+	got, err := announce.Decode(c.announced)
+	if err != nil || got.Module != "mesh-mail" || got.Node != "novox" || !got.Readable("accounts") {
+		t.Fatalf("the announcement is %s (%v)", c.announced, err)
+	}
+}

go/process/state.go

@@ -39,6 +39,9 @@ func (s State) bucket() string { return s.Module + "_" + s.Name }
 func (c *Conn) State(name string) State {
 	bucket := c.account.Module + "_" + name
 	return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) {
+		if err := c.refusedUnannounced(name); err != nil {
+			return nil, err
+		}
 		js, err := jetstream.New(c.nc)
 		if err != nil {
 			return nil, err

go/state.go

@@ -50,6 +50,9 @@ func State(name string) KeptState { return KeptState{name: name} }
 const stateTimeout = 30 * time.Second
 
 func (s KeptState) ask(verb string, params map[string]any) (json.RawMessage, error) {
+	if err := refusedUnannounced(s.name); err != nil {
+		return nil, err
+	}
 	c, err := live()
 	if err != nil {
 		return nil, err

go/stdio.go

@@ -180,6 +180,9 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error {
 	currentMu.Lock()
 	current = c
 	currentMu.Unlock()
+	// What the bundle said it keeps is announced once it serves: the runtime answers the publish on this channel, so
+	// it is asked from beside the loop that reads the answer.
+	go func() { _ = announce1() }()
 	scanner := bufio.NewScanner(in)
 	scanner.Buffer(make([]byte, 1<<20), 1<<24)
 	for scanner.Scan() {

package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "@novox/mesh-sdk",
-  "version": "0.1.14",
+  "version": "0.1.15",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@novox/mesh-sdk",
-      "version": "0.1.14",
+      "version": "0.1.15",
       "devDependencies": {
         "@types/node": "^22.0.0",
         "typescript": "^5.6.0"

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@novox/mesh-sdk",
-  "version": "0.1.14",
+  "version": "0.1.15",
   "description": "The stable spine a Novox Mesh module's own code builds against.",
   "type": "module",
   "exports": {

src/state/index.ts

@@ -6,6 +6,12 @@
 // (`reads`, as `<module>.<name>`), and names each here the way it named it there. Every instance of
 // the owner may write; a reader reads and watches. The runtime that launched the bundle does the work
 // and refuses, with the reason, what the module did not declare.
+//
+// A module also says on the bus, from its code, which states it keeps and which others may read
+// (`keeps`, novox/hq ADR 0495 clause 2): a reader is granted what the owner announced, never what a list
+// or the owner's manifest says (clause 3).
+
+import { emit } from "../events/index.js";
 
 /** One key's current value. */
 export interface StateEntry<T = unknown> {
@@ -61,7 +67,7 @@ export function useStateChannel(c: StateChannel): void {
   channel = c;
 }
 
-function bound(): StateChannel {
+function boundChannel(): StateChannel {
   if (!channel) {
     throw new Error(
       "no runtime to keep state through — a module's state is reached through the runtime that launched " +
@@ -71,8 +77,57 @@ function bound(): StateChannel {
   return channel;
 }
 
+/** One state a module keeps, by its local name, and whether other modules may read it. */
+export interface Kept {
+  readonly name: string;
+  readonly readable?: boolean;
+}
+
+const stateName = /^[a-z0-9][a-z0-9-]*$/;
+let keptNames: Set<string> | undefined;
+
+/**
+ * Say on the bus, as this bundle's module, every state it keeps and which of them others may read
+ * (novox/hq ADR 0495 clause 2; issue 1790): the announcement `states` on the module's own event subject,
+ * `{v: 1, module, node, states}` with the states sorted by name, which the controller keeps and grants
+ * readers from (clause 3). Call it as the bundle is imported. From then on a state of the module's own
+ * that it did not announce is refused at the call, naming it, so code and announcement cannot drift
+ * (clause 5). A state no bucket could be named by, or one given twice, is refused.
+ */
+export async function keeps(states: readonly Kept[]): Promise<void> {
+  const module = process.env.MESH_MODULE || process.env.MESH_SERVED_MODULE || "";
+  if (!stateName.test(module)) {
+    throw new Error(`"${module}" is not a module that keeps state: lower-case letters, digits and hyphens`);
+  }
+  const names = new Set<string>();
+  for (const k of states) {
+    if (!stateName.test(k.name)) {
+      throw new Error(`${module} announces the state "${k.name}", and a state is named locally: lower-case letters, digits and hyphens`);
+    }
+    if (names.has(k.name)) throw new Error(`${module} announces the state "${k.name}" twice`);
+    names.add(k.name);
+  }
+  keptNames = names;
+  const sorted = [...states]
+    .sort((a, b) => (a.name < b.name ? -1 : a.name > b.name ? 1 : 0))
+    .map((k) => (k.readable ? { name: k.name, readable: true } : { name: k.name }));
+  const node = process.env.MESH_NODE;
+  await emit("states", { v: 1, module, ...(node ? { node } : {}), states: sorted });
+}
+
+/** The error of a call on a state of the module's own that `keeps` did not name, before the runtime is asked. */
+function refusedUnannounced(name: string): void {
+  if (!name.includes(".") && keptNames && !keptNames.has(name)) {
+    throw new Error(`the state "${name}" is not among the states this module announced it keeps (keeps)`);
+  }
+}
+
 /** A state this module keeps (by its local name) or reads (`<module>.<name>`). */
 export function state<T = unknown>(name: string): State<T> {
+  const bound = (): StateChannel => {
+    refusedUnannounced(name);
+    return boundChannel();
+  };
   return {
     get: async (key) => (await bound().ask("mesh/state.get", { state: name, key })) as StateEntry<T> | null,
     put: async (key, value) =>

test/fixtures/stdio-keeps.mjs

@@ -0,0 +1,11 @@
+// A launched bundle that says what it keeps (novox/hq ADR 0495 clause 2; issue 1790): it announces its states as it is
+// imported, and a state of its own it did not announce is refused at the call.
+import { keeps, state } from "../../dist/state/index.js";
+import { registerModuleTools } from "../../dist/tools/index.js";
+import { serveRegisteredOverStdio } from "../../dist/stdio/index.js";
+
+await keeps([{ name: "servers", readable: true }, { name: "own" }]);
+registerModuleTools("keeper", () => [
+  { name: "unannounced", description: "reads a state it did not announce", input: {}, run: async () => state("drafts").get("k") },
+]);
+await serveRegisteredOverStdio();

test/stdio.test.ts

@@ -273,3 +273,40 @@ test("topics match as the runtime's do: * one segment, # and ** any number", asy
   assert.ok(topicMatches("alpha.**", "alpha.a.b"));
   assert.ok(!topicMatches("beta.*", "alpha.happened"));
 });
+
+test("a launched bundle announces what it keeps as it is imported, and binds no state it did not announce (ADR 0495, issue 1790)", async () => {
+  const child = spawn(process.execPath, [fileURLToPath(new URL("./fixtures/stdio-keeps.mjs", import.meta.url))],
+    { stdio: ["pipe", "pipe", "inherit"], env: { ...process.env, MESH_SERVED_MODULE: "keeper", MESH_MODULE: "keeper", MESH_NODE: "anchor" } });
+  const seen: any[] = [];
+  let buffered = "";
+  child.stdout.on("data", (d) => {
+    buffered += d.toString();
+    let at: number;
+    while ((at = buffered.indexOf("\n")) >= 0) {
+      seen.push(JSON.parse(buffered.slice(0, at)));
+      buffered = buffered.slice(at + 1);
+    }
+  });
+  const until = async (match: (m: any) => boolean) => {
+    for (let i = 0; i < 150 && !seen.some(match); i++) await new Promise((r) => setTimeout(r, 20));
+    return seen.find(match);
+  };
+  const send = (m: unknown) => child.stdin.write(JSON.stringify(m) + "\n");
+  try {
+    const asked = await until((m) => m.method === "mesh/publish");
+    assert.ok(asked, "the bundle announced nothing as it was imported");
+    assert.equal(asked.params.key, "states");
+    assert.deepEqual(asked.params.body, {
+      v: 1, module: "keeper", node: "anchor", states: [{ name: "own" }, { name: "servers", readable: true }],
+    });
+    send({ jsonrpc: "2.0", id: asked.id, result: {} });
+    send({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-03-26", capabilities: {} } });
+    await until((m) => m.id === 1);
+    send({ jsonrpc: "2.0", id: 2, method: "tools/call", params: { name: "unannounced", arguments: {} } });
+    const answered = await until((m) => m.id === 2);
+    assert.match(JSON.stringify(answered), /drafts/, "a state the bundle did not announce was bound");
+    assert.equal(seen.find((m) => m.method === "mesh/state.get"), undefined, "the runtime was asked for an unannounced state");
+  } finally {
+    child.kill();
+  }
+});