From 096c5562a47eb3531f1add257c0e1aecaf9ae47a Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 4 Oct 2026 02:47:31 +0200 Subject: [PATCH] A module's state through the runtime, in TypeScript and Go (novox/hq ADR 0201) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit state(name) / stdio.State(name): get, put, delete, keys and watch over the runtime's mesh/state verbs. A watch's handler is registered under the id it is asked with, so its current values — which arrive before the watch is answered — and its changes reach it and no other watch. --- README.md | 6 ++ go/README.md | 2 +- go/state.go | 106 ++++++++++++++++++++++++++++++ go/state_test.go | 117 ++++++++++++++++++++++++++++++++++ go/stdio.go | 39 +++++++++++- package-lock.json | 4 +- package.json | 5 +- src/index.ts | 1 + src/state/index.ts | 89 ++++++++++++++++++++++++++ src/stdio/index.ts | 40 +++++++++++- test/fixtures/stdio-state.mjs | 18 ++++++ test/stdio.test.ts | 70 ++++++++++++++++++++ 12 files changed, 488 insertions(+), 9 deletions(-) create mode 100644 go/state.go create mode 100644 go/state_test.go create mode 100644 src/state/index.ts create mode 100644 test/fixtures/stdio-state.mjs diff --git a/README.md b/README.md index 90e5101..f5d7c33 100644 --- a/README.md +++ b/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 `.`): `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. diff --git a/go/README.md b/go/README.md index 6f8d4f8..1d1ae78 100644 --- a/go/README.md +++ b/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 `.` 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 diff --git a/go/state.go b/go/state.go new file mode 100644 index 0000000..aae30bb --- /dev/null +++ b/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 +// `.`) 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 (`.`). +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 +} diff --git a/go/state_test.go b/go/state_test.go new file mode 100644 index 0000000..79c5350 --- /dev/null +++ b/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() +} diff --git a/go/stdio.go b/go/stdio.go index 8e4995a..136309e 100644 --- a/go/stdio.go +++ b/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}}) diff --git a/package-lock.json b/package-lock.json index a99c71d..be7c725 100644 --- a/package-lock.json +++ b/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" diff --git a/package.json b/package.json index e92b3aa..fee8eb5 100644 --- a/package.json +++ b/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" diff --git a/src/index.ts b/src/index.ts index af1a960..39a0633 100644 --- a/src/index.ts +++ b/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"; diff --git a/src/state/index.ts b/src/state/index.ts new file mode 100644 index 0000000..2f7da14 --- /dev/null +++ b/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 `.`), 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 { + 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 { + /** 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 { + /** A key's current value, or null when it has none. */ + get(key: string): Promise | null>; + /** Write a key — the owner's instances only. Resolves to the revision written. */ + put(key: string, value: T): Promise; + /** Remove a key — the owner's instances only. Removing one that is not there is not an error. */ + delete(key: string): Promise; + /** Every key with a current value, sorted. */ + keys(): Promise; + /** + * 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) => Promise | void, opts?: { key?: string }): Promise; +} + +/** 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; + /** Ask for a watch, with `deliver` registered for its changes before the request is sent. */ + watch(params: unknown, deliver: (change: StateChange) => Promise): Promise; +} + +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 (`.`). */ +export function state(name: string): State { + return { + get: async (key) => (await bound().ask("mesh/state.get", { state: name, key })) as StateEntry | 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); + }), + }; +} diff --git a/src/stdio/index.ts b/src/stdio/index.ts index ef5abe3..e99ed2f 100644 --- a/src/stdio/index.ts +++ b/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 void; reject: (e: Error) => void }>(); - const ask = (method: string, params: unknown): Promise => + const askAs = (id: string, method: string, params: unknown): Promise => 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 => 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) => Promise>(); + 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 } | 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}` } }); } diff --git a/test/fixtures/stdio-state.mjs b/test/fixtures/stdio-state.mjs new file mode 100644 index 0000000..9dfdea1 --- /dev/null +++ b/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(); diff --git a/test/stdio.test.ts b/test/stdio.test.ts index b76a5ad..885bd26 100644 --- a/test/stdio.test.ts +++ b/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) => + 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"));