Merge pull request 'A module's state through the runtime, in TypeScript and Go (hq ADR 0201)' (#18) from feat/module-state-on-the-bus into main

README.md

@@ -38,6 +38,12 @@ they call, live in the module.
 The broker client (connect, request/reply, publish/subscribe), the event-consumer a module uses to
 react to mesh events, and the envelope/routing-key conventions. Transport primitives.
 
+### `state` — a module's current state on the bus
+`state(name)` reaches a state the module declared it keeps (`state` in its manifest) or reads
+(`reads`, as `<module>.<name>`): `get`, `put`, `delete`, `keys`, and `watch`, which hands the current
+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).
+
 ### `primitives` — sealing, semver, resolved-env
 Seal/unseal for secrets, semantic version comparison, and the accessor a module uses to read its
 own resolved environment.

go/README.md

@@ -3,7 +3,7 @@
 `stdio.Serve("", tools)` speaks MCP over stdio to the node's tool runtime, as the module the runtime
 names (novox/hq ADR 0188, ADR 0193). A `Tool` is a name, a description, an argument schema and a
 function; calls run concurrently. A tool named `<seat>.<verb>` is the seat's implementation.
-`stdio.Emit(type, body)` publishes an event as the module, `stdio.Subscribe(pattern, handler)` receives the module's events (acknowledged only after the handler returns nil) and `stdio.Ask(key, body)` calls a tool — all through the runtime (novox/hq ADR 0193, 0198). Standard library only.
+`stdio.Emit(type, body)` publishes an event as the module, `stdio.Subscribe(pattern, handler)` receives the module's events (acknowledged only after the handler returns nil) and `stdio.Ask(key, body)` calls a tool, and `stdio.State(name)` gets, puts, deletes, lists and watches the module's declared state — all through the runtime (novox/hq ADR 0193, 0198, 0201). Standard library only.
 
 ## Installing
 

go/state.go

@@ -0,0 +1,106 @@
+package stdio
+
+import (
+	"encoding/json"
+	"time"
+)
+
+// State — the current value of something a module keeps on the bus, which every machine sees,
+// including one that joins later (novox/hq ADR 0201). Not history: that is an event. Never a secret,
+// sealed or not. A module declares what it keeps (`state`) and what it reads (`reads`, as
+// `<module>.<name>`) and names each here as it named it there; the runtime refuses, with the reason,
+// what it did not declare.
+
+// StateEntry is one key's current value.
+type StateEntry struct {
+	Key      string          `json:"key"`
+	Value    json.RawMessage `json:"value"`
+	Revision uint64          `json:"revision"`
+}
+
+// StateChange is one change a watch hands over: a key put or deleted.
+type StateChange struct {
+	State    string          `json:"state"`
+	Key      string          `json:"key"`
+	Op       string          `json:"op"`
+	Value    json.RawMessage `json:"value,omitempty"`
+	Revision uint64          `json:"revision"`
+	// Current is true for a value that was there when the watch began, false for a change since.
+	Current bool `json:"current"`
+}
+
+// KeptState is one state this module keeps, or reads.
+type KeptState struct{ name string }
+
+// State is a state this module keeps (by its local name) or reads (`<module>.<name>`).
+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) {
+	c, err := live()
+	if err != nil {
+		return nil, err
+	}
+	params["state"] = s.name
+	return c.ask("mesh/state."+verb, params, stateTimeout)
+}
+
+// Get is a key's current value, or nil when it has none.
+func (s KeptState) Get(key string) (*StateEntry, error) {
+	raw, err := s.ask("get", map[string]any{"key": key})
+	if err != nil || len(raw) == 0 || string(raw) == "null" {
+		return nil, err
+	}
+	var e StateEntry
+	return &e, json.Unmarshal(raw, &e)
+}
+
+// Put writes a key — the owner's instances only — and returns the revision written.
+func (s KeptState) Put(key string, value any) (uint64, error) {
+	raw, err := s.ask("put", map[string]any{"key": key, "value": value})
+	if err != nil {
+		return 0, err
+	}
+	var put struct {
+		Revision uint64 `json:"revision"`
+	}
+	return put.Revision, json.Unmarshal(raw, &put)
+}
+
+// Delete removes a key — the owner's instances only. A key that was not there is not an error.
+func (s KeptState) Delete(key string) error {
+	_, err := s.ask("delete", map[string]any{"key": key})
+	return err
+}
+
+// Keys is every key with a current value, sorted.
+func (s KeptState) Keys() ([]string, error) {
+	raw, err := s.ask("keys", map[string]any{})
+	if err != nil {
+		return nil, err
+	}
+	var keys []string
+	return keys, json.Unmarshal(raw, &keys)
+}
+
+// Watch hands handle every current value — none that is deleted — and then every change, in order,
+// and returns once the current values have all been handled. key narrows it: a key, with `*` for one
+// name and `**` for the rest; empty is every key. The watch lasts as long as the process.
+func (s KeptState) Watch(key string, handle func(StateChange) error) error {
+	c, err := live()
+	if err != nil {
+		return err
+	}
+	id := c.newID()
+	c.mu.Lock()
+	c.watches[id] = handle
+	c.mu.Unlock()
+	if _, err := c.askAs(id, "mesh/state.watch", map[string]any{"state": s.name, "key": key}, stateTimeout); err != nil {
+		c.mu.Lock()
+		delete(c.watches, id)
+		c.mu.Unlock()
+		return err
+	}
+	return nil
+}

go/state_test.go

@@ -0,0 +1,117 @@
+package stdio
+
+import (
+	"bufio"
+	"encoding/json"
+	"io"
+	"strings"
+	"testing"
+	"time"
+)
+
+// A Go bundle keeps and watches its state through the runtime (novox/hq ADR 0201): the watch's
+// current values are handled before it returns, its changes are named by the watch's id, a change for
+// no watch of this bundle is refused, and put, get and keys are asked of the runtime as the module.
+func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) {
+	t.Setenv(ServedAs, "keeper")
+	inR, inW := io.Pipe()
+	outR, outW := io.Pipe()
+	seen := make(chan string, 16)
+	watched := make(chan error, 1)
+	go func() {
+		_ = ServeOn("", []Tool{
+			{Name: "start", Run: func(map[string]any) (any, error) {
+				go func() {
+					watched <- State("servers").Watch("", func(c StateChange) error {
+						seen <- c.Op + " " + c.Key + " " + string(c.Value)
+						return nil
+					})
+				}()
+				return "started", nil
+			}},
+			{Name: "remember", Run: func(a map[string]any) (any, error) {
+				return State("servers").Put(a["key"].(string), a["value"])
+			}},
+			{Name: "recall", Run: func(a map[string]any) (any, error) { return State("servers").Get(a["key"].(string)) }},
+		}, 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"}})
+	got := map[string]map[string]any{}
+	for len(got) < 2 {
+		m := next()
+		if m["method"] != nil {
+			got[m["method"].(string)] = m
+		} else {
+			got["answer"] = m
+		}
+	}
+	watch := got["mesh/state.watch"]
+	if p := watch["params"].(map[string]any); p["state"] != "servers" || p["key"] != "" {
+		t.Fatalf("asked to watch %v", p)
+	}
+	// The current value, for this watch, before the watch is answered.
+	send(map[string]any{"jsonrpc": "2.0", "id": 50, "method": "mesh/state", "params": map[string]any{"watch": watch["id"],
+		"change": map[string]any{"state": "servers", "key": "all.one", "op": "put", "value": map[string]any{"url": "http://one"}, "current": true}}})
+	if r := next(); r["id"] != float64(50) || r["error"] != nil {
+		t.Fatalf("a current value was answered %v", r)
+	}
+	if s := <-seen; s != `put all.one {"url":"http://one"}` {
+		t.Fatalf("handled %q", s)
+	}
+	select {
+	case err := <-watched:
+		t.Fatalf("the watch returned before it was answered: %v", err)
+	default:
+	}
+	send(map[string]any{"jsonrpc": "2.0", "id": watch["id"], "result": map[string]any{}})
+	if err := <-watched; err != nil {
+		t.Fatal(err)
+	}
+	// A change for a watch this bundle never asked for is refused.
+	send(map[string]any{"jsonrpc": "2.0", "id": 51, "method": "mesh/state", "params": map[string]any{"watch": "bundle-999",
+		"change": map[string]any{"key": "x", "op": "put"}}})
+	if r := next(); r["id"] != float64(51) || !strings.Contains(r["error"].(map[string]any)["message"].(string), "names no watch") {
+		t.Fatalf("a change for no watch was answered %v", r)
+	}
+
+	// Put: asked of the runtime as the module, the revision is the tool's answer.
+	send(map[string]any{"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": map[string]any{"name": "remember",
+		"arguments": map[string]any{"key": "all.two", "value": map[string]any{"url": "http://two"}}}})
+	put := next()
+	if put["method"] != "mesh/state.put" || put["params"].(map[string]any)["key"] != "all.two" || put["params"].(map[string]any)["state"] != "servers" {
+		t.Fatalf("asked %v", put)
+	}
+	send(map[string]any{"jsonrpc": "2.0", "id": put["id"], "result": map[string]any{"revision": 9}})
+	if r := next(); r["id"] != float64(2) || r["result"].(map[string]any)["content"].([]any)[0].(map[string]any)["text"] != "9" {
+		t.Fatalf("the put answered %v", r)
+	}
+	// Get of a key with no value is nil.
+	send(map[string]any{"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": map[string]any{"name": "recall",
+		"arguments": map[string]any{"key": "nothing"}}})
+	get := next()
+	send(map[string]any{"jsonrpc": "2.0", "id": get["id"], "result": nil})
+	if r := next(); r["id"] != float64(3) || r["result"].(map[string]any)["content"].([]any)[0].(map[string]any)["text"] != "null" {
+		t.Fatalf("the get answered %v", r)
+	}
+	_ = inW.Close()
+}

go/stdio.go

@@ -60,6 +60,8 @@ type channel struct {
 	next      int
 	waiting   map[string]chan message
 	listeners []listener
+	// watches are this bundle's state watches by the id each was asked with (ADR 0201).
+	watches map[string]func(StateChange) error
 }
 
 type listener struct {
@@ -85,9 +87,20 @@ func (c *channel) say(m map[string]any) {
 
 // ask sends the runtime a request and waits for its answer.
 func (c *channel) ask(method string, params any, timeout time.Duration) (json.RawMessage, error) {
+	return c.askAs(c.newID(), method, params, timeout)
+}
+
+func (c *channel) newID() string {
 	c.mu.Lock()
+	defer c.mu.Unlock()
 	c.next++
-	id := "bundle-" + strconv.Itoa(c.next)
+	return "bundle-" + strconv.Itoa(c.next)
+}
+
+// askAs is ask under an id the caller chose, for a request whose id something is registered under
+// before it is sent.
+func (c *channel) askAs(id, method string, params any, timeout time.Duration) (json.RawMessage, error) {
+	c.mu.Lock()
 	answer := make(chan message, 1)
 	c.waiting[id] = answer
 	c.mu.Unlock()
@@ -129,7 +142,7 @@ 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{}}
+	c := &channel{enc: json.NewEncoder(out), waiting: map[string]chan message{}, watches: map[string]func(StateChange) error{}}
 	currentMu.Lock()
 	current = c
 	currentMu.Unlock()
@@ -233,6 +246,28 @@ 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.Envelope, matching)
+		case "mesh/state":
+			// One change to a state this bundle watches (ADR 0201), named by the watch's id, answered
+			// once its handler has taken it. Beside the loop, like a tool call.
+			var p struct {
+				Watch  string       `json:"watch"`
+				Change *StateChange `json:"change"`
+			}
+			_ = json.Unmarshal(m.Params, &p)
+			c.mu.Lock()
+			handle := c.watches[p.Watch]
+			c.mu.Unlock()
+			if handle == nil || p.Change == nil {
+				c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/state names no watch of this bundle"}})
+				continue
+			}
+			go func(id any, change StateChange) {
+				if err := handle(change); 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.Change)
 		default:
 			if id != nil {
 				c.say(map[string]any{"id": id, "error": map[string]any{"code": -32601, "message": "no " + m.Method}})

package-lock.json

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

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@novox/mesh-sdk",
-  "version": "0.1.6",
+  "version": "0.1.7",
   "description": "The stable spine a Novox Mesh module's own code builds against.",
   "type": "module",
   "exports": {
@@ -11,7 +11,8 @@
     "./messaging": "./dist/messaging/index.js",
     "./events": "./dist/events/index.js",
     "./primitives": "./dist/primitives/index.js",
-    "./stdio": "./dist/stdio/index.js"
+    "./stdio": "./dist/stdio/index.js",
+    "./state": "./dist/state/index.js"
   },
   "files": [
     "dist"

src/index.ts

@@ -10,3 +10,4 @@ export * as provisioner from "./provisioner/index.js";
 export * as messaging from "./messaging/index.js";
 export * as events from "./events/index.js";
 export * as primitives from "./primitives/index.js";
+export * as state from "./state/index.js";

src/state/index.ts

@@ -0,0 +1,89 @@
+// State — the current value of something a module keeps on the bus, which every machine sees,
+// including one that joins later (novox/hq ADR 0201). Not history: that is an event. Never a secret,
+// sealed or not: a bucket is kept, and a machine joining a year later reads it whole.
+//
+// A module declares the state it keeps (`state` in its manifest) and the state of others it reads
+// (`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.
+
+/** One key's current value. */
+export interface StateEntry<T = unknown> {
+  readonly key: string;
+  readonly value: T;
+  /** Increases with every write to the bucket; a later value has a larger one. */
+  readonly revision: number;
+}
+
+/** One change a watch hands over. */
+export interface StateChange<T = unknown> {
+  /** The state, as the module named it. */
+  readonly state: string;
+  readonly key: string;
+  readonly op: "put" | "delete";
+  /** The new value; absent for a delete. */
+  readonly value?: T;
+  readonly revision: number;
+  /** True for a value that was there when the watch began, false for a change since. */
+  readonly current: boolean;
+}
+
+/** One state a module keeps or reads. */
+export interface State<T = unknown> {
+  /** A key's current value, or null when it has none. */
+  get(key: string): Promise<StateEntry<T> | null>;
+  /** Write a key — the owner's instances only. Resolves to the revision written. */
+  put(key: string, value: T): Promise<number>;
+  /** Remove a key — the owner's instances only. Removing one that is not there is not an error. */
+  delete(key: string): Promise<void>;
+  /** Every key with a current value, sorted. */
+  keys(): Promise<string[]>;
+  /**
+   * Hand `handler` every current value — none that is deleted — and then every change, in order.
+   * Resolves once the current values have all been handled, so code after it has the whole state.
+   * `key` narrows it: a key, with `*` for one name and `**` for the rest. The watch lasts as long as
+   * the process; a bundle that starts again watches again as its code runs again.
+   */
+  watch(handler: (change: StateChange<T>) => Promise<void> | void, opts?: { key?: string }): Promise<void>;
+}
+
+/** What the state surface needs of the channel to the runtime; the stdio session provides it. */
+export interface StateChannel {
+  ask(method: string, params: unknown): Promise<unknown>;
+  /** Ask for a watch, with `deliver` registered for its changes before the request is sent. */
+  watch(params: unknown, deliver: (change: StateChange<unknown>) => Promise<void>): Promise<void>;
+}
+
+let channel: StateChannel | undefined;
+
+/** Bind the channel. The stdio session calls this when the runtime launched the process. */
+export function useStateChannel(c: StateChannel): void {
+  channel = c;
+}
+
+function bound(): StateChannel {
+  if (!channel) {
+    throw new Error(
+      "no runtime to keep state through — a module's state is reached through the runtime that launched " +
+        "its bundle (novox/hq ADR 0201)",
+    );
+  }
+  return channel;
+}
+
+/** A state this module keeps (by its local name) or reads (`<module>.<name>`). */
+export function state<T = unknown>(name: string): State<T> {
+  return {
+    get: async (key) => (await bound().ask("mesh/state.get", { state: name, key })) as StateEntry<T> | null,
+    put: async (key, value) =>
+      ((await bound().ask("mesh/state.put", { state: name, key, value })) as { revision: number }).revision,
+    delete: async (key) => {
+      await bound().ask("mesh/state.delete", { state: name, key });
+    },
+    keys: async () => (await bound().ask("mesh/state.keys", { state: name })) as string[],
+    watch: async (handler, opts = {}) =>
+      bound().watch({ state: name, key: opts.key ?? "" }, async (change) => {
+        await handler(change as StateChange<T>);
+      }),
+  };
+}

src/stdio/index.ts

@@ -15,6 +15,7 @@
 import type { ToolDefinition } from "../contracts/index.js";
 import { collectTools } from "../tools/index.js";
 import { useBroker, type Broker, type Envelope } from "../messaging/index.js";
+import { useStateChannel, type StateChange } from "../state/index.js";
 
 /** The protocol version this speaks; the runtime says the same. */
 export const PROTOCOL = "2025-03-26";
@@ -63,12 +64,30 @@ function startSession(): Session {
   };
   let nextOut = 1;
   const awaiting = new Map<string, { resolve: (v: unknown) => void; reject: (e: Error) => void }>();
-  const ask = (method: string, params: unknown): Promise<unknown> =>
+  const askAs = (id: string, method: string, params: unknown): Promise<unknown> =>
     new Promise((resolve, reject) => {
-      const id = `bundle-${nextOut++}`;
       awaiting.set(id, { resolve, reject });
       say({ jsonrpc: "2.0", id, method, params });
     });
+  const ask = (method: string, params: unknown): Promise<unknown> => askAs(`bundle-${nextOut++}`, method, params);
+
+  // The module's state (ADR 0201): each watch's changes arrive as `mesh/state` requests naming the
+  // watch by the id it was asked with, so its handler is registered under that id before it is asked
+  // — the current values arrive before the watch itself is answered.
+  const watches = new Map<string, (change: StateChange<unknown>) => Promise<void>>();
+  useStateChannel({
+    ask,
+    watch: async (params, deliver) => {
+      const id = `bundle-${nextOut++}`;
+      watches.set(id, deliver);
+      try {
+        await askAs(id, "mesh/state.watch", params);
+      } catch (err) {
+        watches.delete(id);
+        throw err;
+      }
+    },
+  });
 
   // What the module's code reacts to: every handler, matched here against each event the runtime
   // delivers. The runtime binds the module's one durable consumer and hands every message it reads to
@@ -156,6 +175,23 @@ function startSession(): Session {
         }
         return;
       }
+      case "mesh/state": {
+        // One change to a state this bundle watches (ADR 0201), answered once its handler has taken
+        // it. The runtime delivers a watch's changes one at a time, in order.
+        const p = request.params as { watch?: string; change?: StateChange<unknown> } | undefined;
+        const deliver = p?.watch !== undefined ? watches.get(String(p.watch)) : undefined;
+        if (!deliver || !p?.change) {
+          say({ jsonrpc: "2.0", id: request.id, error: { code: -32602, message: "mesh/state names no watch of this bundle" } });
+          return;
+        }
+        try {
+          await deliver(p.change);
+          say({ jsonrpc: "2.0", id: request.id, result: {} });
+        } catch (err) {
+          say({ jsonrpc: "2.0", id: request.id, error: { code: -32000, message: err instanceof Error ? err.message : String(err) } });
+        }
+        return;
+      }
       default:
         if (!notification) say({ jsonrpc: "2.0", id: request.id, error: { code: -32601, message: `no ${request.method}` } });
     }

test/fixtures/stdio-state.mjs

@@ -0,0 +1,18 @@
+// A launched bundle that keeps state through its runtime (novox/hq ADR 0201): it watches its servers as
+// it is imported, before it says what it serves, and its tools write, read and list them.
+import { state } from "../../dist/state/index.js";
+import { registerModuleTools } from "../../dist/tools/index.js";
+import { serveRegisteredOverStdio } from "../../dist/stdio/index.js";
+
+const servers = state("servers");
+await servers.watch((c) => {
+  process.stderr.write(`${c.current ? "was" : "now"} ${c.op} ${c.key} ${JSON.stringify(c.value ?? null)}\n`);
+  if (c.value && c.value.fail) throw new Error("not this one");
+});
+process.stderr.write("watching\n");
+registerModuleTools("keeper", () => [
+  { name: "remember", description: "put", input: {}, run: async (a) => servers.put(a.key, a.value) },
+  { name: "recall", description: "get", input: {}, run: async (a) => servers.get(a.key) },
+  { name: "list", description: "keys", input: {}, run: async () => state("licence-manager.bindings").keys() },
+]);
+await serveRegisteredOverStdio();

test/stdio.test.ts

@@ -160,6 +160,76 @@ test("a launched bundle subscribes as it is imported, handles events the runtime
   }
 });
 
+test("a launched bundle watches its state as it is imported, has the current values before it goes on, and keeps state through the runtime (ADR 0201)", async () => {
+  const child = spawn(process.execPath, [fileURLToPath(new URL("./fixtures/stdio-state.mjs", import.meta.url))],
+    { stdio: ["pipe", "pipe", "pipe"], env: { ...process.env, MESH_SERVED_MODULE: "keeper" } });
+  const seen: any[] = [];
+  let buffered = "";
+  let said = "";
+  child.stderr.on("data", (d) => (said += d.toString()));
+  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");
+  const change = (id: number, watch: string, c: Record<string, unknown>) =>
+    send({ jsonrpc: "2.0", id, method: "mesh/state", params: { watch, change: { state: "servers", revision: id, ...c } } });
+  try {
+    // The watch comes first, from the import; the current values arrive before it is answered.
+    const watch = await until((m) => m.method === "mesh/state.watch");
+    assert.deepEqual(watch.params, { state: "servers", key: "" });
+    change(100, watch.id, { key: "all.one", op: "put", value: { url: "http://one" }, current: true });
+    assert.deepEqual((await until((m) => m.id === 100)).result, {});
+    assert.match(said, /was put all\.one \{"url":"http:\/\/one"\}/);
+    assert.doesNotMatch(said, /watching/, "the import went on before the watch was answered");
+    send({ jsonrpc: "2.0", id: watch.id, result: {} });
+    await until(() => said.includes("watching"));
+    assert.match(said, /watching/);
+
+    // A change for a watch this bundle never asked for is refused, not handed to somebody else.
+    send({ jsonrpc: "2.0", id: 101, method: "mesh/state", params: { watch: "bundle-999", change: { key: "x", op: "put" } } });
+    assert.match((await until((m) => m.id === 101)).error.message, /names no watch/);
+    // A change the handler fails is answered as an error.
+    change(102, watch.id, { key: "all.two", op: "put", value: { fail: true }, current: false });
+    assert.match((await until((m) => m.id === 102)).error.message, /not this one/);
+    change(103, watch.id, { key: "all.one", op: "delete", current: false });
+    assert.deepEqual((await until((m) => m.id === 103)).result, {});
+    assert.match(said, /now delete all\.one null/);
+
+    send({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-03-26", capabilities: {} } });
+    assert.equal((await until((m) => m.id === 1)).result.serverInfo.name, "keeper");
+
+    // A tool writes: the runtime is asked, and the revision it answers is the tool's.
+    send({ jsonrpc: "2.0", id: 2, method: "tools/call", params: { name: "remember", arguments: { key: "all.three", value: { url: "http://three" } } } });
+    const put = await until((m) => m.method === "mesh/state.put");
+    assert.deepEqual(put.params, { state: "servers", key: "all.three", value: { url: "http://three" } });
+    send({ jsonrpc: "2.0", id: put.id, result: { revision: 7 } });
+    assert.equal(JSON.parse((await until((m) => m.id === 2)).result.content[0].text), 7);
+    // A read of a key with no value is null; a refusal is the tool's error, in the runtime's words.
+    send({ jsonrpc: "2.0", id: 3, method: "tools/call", params: { name: "recall", arguments: { key: "nothing" } } });
+    const get = await until((m) => m.method === "mesh/state.get");
+    send({ jsonrpc: "2.0", id: get.id, result: null });
+    assert.equal(JSON.parse((await until((m) => m.id === 3)).result.content[0].text), null);
+    send({ jsonrpc: "2.0", id: 4, method: "tools/call", params: { name: "list", arguments: {} } });
+    const keys = await until((m) => m.method === "mesh/state.keys");
+    assert.deepEqual(keys.params, { state: "licence-manager.bindings" });
+    send({ jsonrpc: "2.0", id: keys.id, error: { code: -32000, message: "keeper keeps and reads no state called" } });
+    const refused = await until((m) => m.id === 4);
+    assert.equal(refused.result.isError, true);
+    assert.match(refused.result.content[0].text, /keeps and reads no state called/);
+  } finally {
+    child.kill();
+  }
+});
+
 test("topics match as the runtime's do: * one segment, # and ** any number", async () => {
   const { topicMatches } = await import("../dist/stdio/index.js");
   assert.ok(topicMatches("alpha.happened", "alpha.happened"));