From 33d136e70227b9a574569f7a598431a805b28478 Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Fri, 16 Oct 2026 09:41:17 +0000 Subject: [PATCH] Feature 57 (config and provisioning): release mesh-tools (the provider's StateBus, the SDK pin, templates), reviewed --- merge-check.sh | 5 +- node-tools/cmd/node-tools/main.go | 6 + node-tools/go.mod | 4 +- node-tools/go.sum | 4 +- node-tools/internal/bus/provision.go | 98 ++++++++ .../internal/launch/event_answer_test.go | 4 + node-tools/internal/launch/launch.go | 29 +++ node-tools/internal/launch/provision_test.go | 107 +++++++++ node-tools/internal/provide/provide.go | 132 +++++++++++ node-tools/internal/provide/provide_test.go | 149 ++++++++++++ node-tools/internal/runtime/provision.go | 216 ++++++++++++++++++ node-tools/internal/runtime/provision_test.go | 203 ++++++++++++++++ node-tools/internal/runtime/runtime.go | 3 + node-tools/module.json | 11 +- 14 files changed, 963 insertions(+), 8 deletions(-) create mode 100644 node-tools/internal/bus/provision.go create mode 100644 node-tools/internal/launch/provision_test.go create mode 100644 node-tools/internal/provide/provide.go create mode 100644 node-tools/internal/provide/provide_test.go create mode 100644 node-tools/internal/runtime/provision.go create mode 100644 node-tools/internal/runtime/provision_test.go diff --git a/merge-check.sh b/merge-check.sh index 5d7e7e3..46acd95 100644 --- a/merge-check.sh +++ b/merge-check.sh @@ -37,8 +37,9 @@ if ! command -v node >/dev/null 2>&1 || [ ! -d node_modules/@novox/mesh-sdk ]; t # And what of the runtime launches no bundle: how a bundle's seat traffic reaches the bus, what its # environment holds, what a runtime on a module's own account serves (novox/hq ADR 0259), and that a write # by compare-and-set loses no update under contention (novox/hq research 042), and a module's log verbs - # (novox/hq ADR 0297), and a key's history (novox/hq ADR 0307). - runtimeless='^(TestALogAnswersEachVerbOnTheBus|TestALogPageIsBoundedByItsBytes|TestAnAppendAtAStaleExpectIsRefusedAsALostUpdate|TestALogNotIssuedIsRefusedWithTheReason|TestWhatALogCannotHoldIsRefusedWithTheReason|TestALogRefusesAValueNamingACredential|TestAKeysHistoryIsEveryValueKeptOldestFirst|TestAKeysHistoryIsRefusedToAModuleNotIssuedTheState|TestARuntimeOnAModulesOwnAccountServesThatModuleAlone|TestNoBusWordReachesABundle|TestAToolCallNeverReachesASeatsEventAcceptOrProof|TestABundlesSeatTrafficGoesThroughItsModulesBus|TestALostCompareAndSetCarriesItsCode|TestTwoWritersIncrementingOneCounterOnTheBusLoseNoUpdate|TestAStaleWriteIsRefusedAndWritesNothing|TestACallOnTheCallerNamedSubjectReachesTheModuleWithItsCaller|TestTheSubjectThatNamesNoCallerStillWorksAndNamesNone)$' + # (novox/hq ADR 0297), and a key's history (novox/hq ADR 0307), and a module's values asked of its machine's + # provisioner client (novox/hq ADR 0370). + runtimeless='^(TestALogAnswersEachVerbOnTheBus|TestALogPageIsBoundedByItsBytes|TestAnAppendAtAStaleExpectIsRefusedAsALostUpdate|TestALogNotIssuedIsRefusedWithTheReason|TestWhatALogCannotHoldIsRefusedWithTheReason|TestALogRefusesAValueNamingACredential|TestAKeysHistoryIsEveryValueKeptOldestFirst|TestAKeysHistoryIsRefusedToAModuleNotIssuedTheState|TestARuntimeOnAModulesOwnAccountServesThatModuleAlone|TestNoBusWordReachesABundle|TestAToolCallNeverReachesASeatsEventAcceptOrProof|TestABundlesSeatTrafficGoesThroughItsModulesBus|TestALostCompareAndSetCarriesItsCode|TestTwoWritersIncrementingOneCounterOnTheBusLoseNoUpdate|TestAStaleWriteIsRefusedAndWritesNothing|TestACallOnTheCallerNamedSubjectReachesTheModuleWithItsCaller|TestTheSubjectThatNamesNoCallerStillWorksAndNamesNone|TestABundleAsksForItsValuesAsItsOwnModule|TestABundlesWatchIsHandedTheValueAndEachChange)$' fi if command -v gcc >/dev/null 2>&1; then CGO_ENABLED=1 go test -race -count=1 $packages diff --git a/node-tools/cmd/node-tools/main.go b/node-tools/cmd/node-tools/main.go index 8a851bd..2eef656 100644 --- a/node-tools/cmd/node-tools/main.go +++ b/node-tools/cmd/node-tools/main.go @@ -24,6 +24,7 @@ import ( "github.com/novox/mesh-tools/node-tools/internal/alive" "github.com/novox/mesh-tools/node-tools/internal/bus" "github.com/novox/mesh-tools/node-tools/internal/console" + "github.com/novox/mesh-tools/node-tools/internal/provide" "github.com/novox/mesh-tools/node-tools/internal/runtime" ) @@ -62,6 +63,10 @@ func main() { stopSaying = alive.Keep(conn, cred.Node, alive.Every, log.Printf) } + // As the machine's runtime it is also the provider of `mcp-endpoint`: it answers its consumers through the + // SDK's helper wherever it runs, whether or not the module is handed (novox/hq ADR 0349 clause 12, issue 791). + stopProviding := provide.Start(conn, cred.Module, cred.Node, time.Minute, log.Printf) + listen := os.Getenv("MESH_CONSOLE_LISTEN") if listen == "" && cred.Module == runtimeModule { listen = consoleListen @@ -84,6 +89,7 @@ func main() { signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT) <-signals stopSaying() + stopProviding() stop() if up != nil { _ = up.Close() diff --git a/node-tools/go.mod b/node-tools/go.mod index 4395904..a1b54d7 100644 --- a/node-tools/go.mod +++ b/node-tools/go.mod @@ -8,15 +8,15 @@ require ( ) require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect github.com/google/go-tpm v0.9.8 // indirect github.com/klauspost/compress v1.20.0 // indirect github.com/minio/highwayhash v1.0.4 // indirect github.com/nats-io/jwt/v2 v2.8.1 // indirect - github.com/nats-io/nats-server/v2 v2.11.17 // indirect + github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect - go.uber.org/automaxprocs v1.6.0 // indirect golang.org/x/crypto v0.57.0 // indirect golang.org/x/time v0.15.0 // indirect ) diff --git a/node-tools/go.sum b/node-tools/go.sum index 2274dd5..808e3ad 100644 --- a/node-tools/go.sum +++ b/node-tools/go.sum @@ -1,3 +1,5 @@ +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19 h1:mAMt3KFmrHzDiBv4MLL94opWcG9y73ISMrBUyisapQE= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261015210729-d3e182503c19/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= @@ -16,8 +18,6 @@ github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= -go.uber.org/automaxprocs v1.6.0 h1:O3y2/QNTOdbF+e/dpXNNW7Rx2hZ4sTIPyybbxyNqTUs= -go.uber.org/automaxprocs v1.6.0/go.mod h1:ifeIMSnPZuznNm6jmdzmU3/bfk01Fe2fotchwEFJ8r8= golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= diff --git a/node-tools/internal/bus/provision.go b/node-tools/internal/bus/provision.go new file mode 100644 index 0000000..2e37576 --- /dev/null +++ b/node-tools/internal/bus/provision.go @@ -0,0 +1,98 @@ +package bus + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/nats-io/nats.go" +) + +// A module's code asks its machine's provisioner client for its values at runtime (novox/hq ADR 0349 clause 14 as +// ADR 0370 changes it, issue 685). +// +// While a module's code runs here, its asks reach the bus under this runtime's account, and the bus proves only +// that account. So the runtime asks its own machine's client on ProvisionAskSubject — which the controller grants +// this machine's tool runner alone to publish (ADR 0370 §3) — naming the module itself, from the bundle that asks +// and never from the module's code or arguments (§4). The client answers that module's config values, for what its +// manifest declares, and no secret: the vault hands a secret only to a caller the bus proves (§2). + +// ProvisionAskSubject is where this machine's tool runner asks its provisioner client (mesh-controller +// internal/broker ProvisionerAskSubject): `prov.ask.`. +func ProvisionAskSubject(node string) string { return "prov.ask." + node } + +// ProvisionChangedSubjects is every module's word of what moved on a machine (mesh-controller internal/broker +// ProvisionerChangedSubject): `prov.changed..`, a module's name dots and all. +func ProvisionChangedSubjects(node string) string { return "prov.changed." + node + ".>" } + +// ProvisionTimeout bounds one ask of the client; a variable so a test need not wait. +var ProvisionTimeout = 10 * time.Second + +// ProvisionChanged is what the client says moved of one module: the kind, and the key, or every key of the kind when +// empty. Never the value. +type ProvisionChanged struct { + Kind string `json:"kind"` + Key string `json:"key,omitempty"` +} + +// ProvisionRefused is the client's refusal, in its own words: a key the module does not declare, a module not placed +// on the machine, a secret, a kind it does not answer yet. +type ProvisionRefused struct{ Reason string } + +func (r *ProvisionRefused) Error() string { return r.Reason } + +// AskProvisioner asks this machine's provisioner client for one value of module: kind `config` or `bound`, and the +// key. The module is the caller's to name from the bundle it runs; this never takes it from what a bundle sent. +func (c *Conn) AskProvisioner(module, kind, key string) (json.RawMessage, error) { + if c.node == "" { + return nil, errors.New("this runtime names no machine, so it has no provisioner client to ask") + } + body, err := json.Marshal(map[string]string{"module": module, "kind": kind, "key": key}) + if err != nil { + return nil, err + } + msg, err := c.nc.Request(ProvisionAskSubject(c.node), body, ProvisionTimeout) + if errors.Is(err, nats.ErrNoResponders) { + return nil, fmt.Errorf("%s's provisioner client does not answer: it is not running there, or the bus does "+ + "not let %s ask it", c.node, c.cred.User) + } + if err != nil { + return nil, fmt.Errorf("asking %s's provisioner client for %s's %s %s: %w", c.node, module, kind, key, err) + } + var answered struct { + Value json.RawMessage `json:"value"` + Error string `json:"error"` + } + if err := json.Unmarshal(msg.Data, &answered); err != nil { + return nil, fmt.Errorf("%s's provisioner client answered something that is not its answer: %w", c.node, err) + } + if answered.Error != "" { + return nil, &ProvisionRefused{Reason: answered.Error} + } + if len(answered.Value) == 0 { + return nil, fmt.Errorf("%s's provisioner client answered no value and no reason", c.node) + } + return answered.Value, nil +} + +// OnProvisionChanged hands told every word of this machine's provisioner client of what moved, by module, until +// stopped. +func (c *Conn) OnProvisionChanged(told func(module string, changed ProvisionChanged)) (func(), error) { + if c.node == "" { + return nil, errors.New("this runtime names no machine, so it hears no provisioner client") + } + prefix := "prov.changed." + c.node + "." + sub, err := c.nc.Subscribe(ProvisionChangedSubjects(c.node), func(m *nats.Msg) { + var changed ProvisionChanged + if json.Unmarshal(m.Data, &changed) != nil || changed.Kind == "" { + return + } + told(strings.TrimPrefix(m.Subject, prefix), changed) + }) + if err != nil { + return nil, err + } + return func() { _ = sub.Unsubscribe() }, c.nc.Flush() +} diff --git a/node-tools/internal/launch/event_answer_test.go b/node-tools/internal/launch/event_answer_test.go index e731154..05d8594 100644 --- a/node-tools/internal/launch/event_answer_test.go +++ b/node-tools/internal/launch/event_answer_test.go @@ -44,6 +44,10 @@ func (b *subscribing) Watch(json.RawMessage, func(json.RawMessage) error) (func( func (b *subscribing) Seat(string, json.RawMessage, func(string, any) (json.RawMessage, error)) (json.RawMessage, error) { return nil, nil } +func (b *subscribing) Provision(string, json.RawMessage) (json.RawMessage, error) { return nil, nil } +func (b *subscribing) ProvisionWatch(json.RawMessage, func(json.RawMessage) error) (func(), error) { + return func() {}, nil +} func (b *subscribing) Subscribe(d func(json.RawMessage) error) error { b.mu.Lock() defer b.mu.Unlock() diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index 6e96aea..5609b6f 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -83,6 +83,11 @@ type Bus interface { // `take` and `answer` bind once and hand each message or proof on through handOn, to whichever child // of the module asked last — a restarted child asks again as its code runs again. Seat(verb string, params json.RawMessage, handOn func(method string, params any) (json.RawMessage, error)) (json.RawMessage, error) + // Provision answers a bundle's `get` of one of its module's values at runtime, and ProvisionWatch hands deliver + // that value now and each change after, as the module whose bundle asks and never a module the bundle names + // (novox/hq ADR 0349 §14, ADR 0370 §4): its config, and its bindings' values that are not secret. + Provision(verb string, params json.RawMessage) (json.RawMessage, error) + ProvisionWatch(params json.RawMessage, deliver func(change json.RawMessage) error) (stop func(), err error) } // EventTimeout bounds how long a bundle has to handle one event before it is offered again. @@ -307,6 +312,30 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an if answered, err = mesh.Log(strings.TrimPrefix(m.Method, "mesh/log."), m.Params); err == nil { result = answered } + case "mesh/provision.get": + var answered json.RawMessage + if answered, err = mesh.Provision("get", m.Params); err == nil { + result = answered + } + case "mesh/provision.watch": + // As a state's watch: each change asked of this child in order, naming the watch by the + // id the child asked it with, and answered once the current value is handed over. + var stop func() + watch := m.ID + stop, err = mesh.ProvisionWatch(m.Params, func(change json.RawMessage) error { + _, err := c.ask(module, "mesh/provision", map[string]any{"watch": watch, "change": change}, EventTimeout) + return err + }) + if err == nil { + c.mu.Lock() + if c.why != nil { + c.mu.Unlock() + stop() + } else { + c.watches = append(c.watches, stop) + c.mu.Unlock() + } + } case "mesh/state.watch": // Each change is asked of this child, in order, naming the watch it is for by // the id the child asked it with — two watches of one state are two handlers; diff --git a/node-tools/internal/launch/provision_test.go b/node-tools/internal/launch/provision_test.go new file mode 100644 index 0000000..44533ec --- /dev/null +++ b/node-tools/internal/launch/provision_test.go @@ -0,0 +1,107 @@ +package launch + +import ( + "encoding/json" + "os" + "path/filepath" + "slices" + "strings" + "sync" + "testing" + "time" +) + +// A bundle that asks for one of its values and watches it, and writes each change it is handed to $OUT. +const provisionBundle = `#!/bin/sh +while IFS= read -r line; do + id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\)[,}].*/\1/p') + case "$line" in + *'"method":"initialize"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;; + *'"method":"tools/list"'*) + printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id" + printf '{"jsonrpc":"2.0","id":"g1","method":"mesh/provision.get","params":{"kind":"config","key":"role"}}\n' + printf '{"jsonrpc":"2.0","id":"w1","method":"mesh/provision.watch","params":{"kind":"config","key":"role"}}\n' ;; + *'"method":"mesh/provision"'*) + printf '%s\n' "$line" >> "$OUT" + printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;; + *'"id":"g1"'*) printf '%s\n' "$line" >> "$OUT" ;; + esac +done +` + +type provisioning struct { + subscribing + mu sync.Mutex + asked []string +} + +func (b *provisioning) Provision(verb string, params json.RawMessage) (json.RawMessage, error) { + b.mu.Lock() + defer b.mu.Unlock() + b.asked = append(b.asked, verb+" "+string(params)) + return json.RawMessage(`"the build machine"`), nil +} + +func (b *provisioning) ProvisionWatch(params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) { + b.mu.Lock() + b.asked = append(b.asked, "watch "+string(params)) + b.mu.Unlock() + return func() {}, deliver(json.RawMessage(`{"kind":"config","key":"role","value":"the build machine"}`)) +} + +// novox/hq ADR 0349 §14, ADR 0370 §4: a bundle's ask for its values reaches the runtime's bus from its own channel, +// answered there, and a watch's change is handed to it naming the watch it asked. +func TestABundleAsksForItsValuesAndIsHandedItsWatchsChanges(t *testing.T) { + entry := filepath.Join(t.TempDir(), "bundle") + if err := os.WriteFile(entry, []byte(provisionBundle), 0o755); err != nil { + t.Fatal(err) + } + out := filepath.Join(t.TempDir(), "out") + b := &provisioning{} + l, err := Start("claude-code", entry, append(os.Environ(), "OUT="+out), b, t.Logf) + if err != nil { + t.Fatal(err) + } + t.Cleanup(l.Stop) + deadline := time.Now().Add(10 * time.Second) + for { + said, _ := os.ReadFile(out) + if strings.Count(string(said), "\n") >= 2 { + // Each line as the bundle read it, by what it is: the get's answer, and the watch's change. + got := map[string]string{} + for _, line := range strings.Split(strings.TrimSpace(string(said)), "\n") { + var m struct { + ID any `json:"id"` + Method string `json:"method"` + Result json.RawMessage `json:"result"` + Params json.RawMessage `json:"params"` + } + if err := json.Unmarshal([]byte(line), &m); err != nil { + t.Fatalf("the bundle read %s", line) + } + if m.Method != "" { + got[m.Method] = string(m.Params) + } else { + got["answer"] = string(m.Result) + } + } + if got["answer"] != `"the build machine"` { + t.Errorf("the get was answered %s", got["answer"]) + } + if got["mesh/provision"] != `{"change":{"kind":"config","key":"role","value":"the build machine"},"watch":"w1"}` { + t.Errorf("the watch was handed %s", got["mesh/provision"]) + } + break + } + if time.Now().After(deadline) { + t.Fatalf("the bundle was handed %q", said) + } + time.Sleep(20 * time.Millisecond) + } + b.mu.Lock() + defer b.mu.Unlock() + slices.Sort(b.asked) + if len(b.asked) != 2 || b.asked[0] != `get {"kind":"config","key":"role"}` || b.asked[1] != `watch {"kind":"config","key":"role"}` { + t.Errorf("the runtime's bus was asked %q", b.asked) + } +} diff --git a/node-tools/internal/provide/provide.go b/node-tools/internal/provide/provide.go new file mode 100644 index 0000000..068ac59 --- /dev/null +++ b/node-tools/internal/provide/provide.go @@ -0,0 +1,132 @@ +// Package provide is the runtime's half of node-tools being a provider of `mcp-endpoint`: it runs the mesh SDK's +// provider helper, so a consumer handed to the provisioner (claude-code's `${bound:mcp-endpoint:port}`) is answered +// from an entry node-tools writes in its own bucket, not from the controller's binding file (novox/hq ADR 0349 +// clause 12, issue 791). +// +// The offer carries no credential: the console's authority is the machine's login (ADR 0152, 0175), so the helper +// reads nothing from the vault, needs no consumers' file and the entry names none. node-tools is not marked moved: +// its manifest says `"answered-by": "helper"` for the provision, and the helper runs whenever the runtime does +// (issue 791), as route-proxy's does (issue 792). +package provide + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "git.novox.be/novox/mesh-sdk/go/provider" + + "github.com/novox/mesh-tools/node-tools/internal/bus" +) + +const ( + // Module is the provider: the runtime's own module. + Module = "node-tools" + // Provision is what node-tools provides, as its manifest says. + Provision = "mcp-endpoint" + // ServedPort is the port its manifest serves (`serves.mcp-endpoint.port`): where the console listens. + ServedPort = 4270 +) + +// Offers is what node-tools serves, per provision. +func Offers() map[string]provider.Offer { + return map[string]provider.Offer{Provision: { + Credential: provider.NoCredential, + Serves: map[string]any{"port": ServedPort}, + }} +} + +// busOf is the helper's bus over the runtime's own state: the provider's bucket is its state `bound`, and the +// controller's assignments are the state it reads, `mesh-controller.assignments` (both declared in the manifest). +type busOf struct{ conn *bus.Conn } + +func (b busOf) state(bucket string) (string, error) { + switch bucket { + case provider.BucketOf(Module): + return "bound", nil + case provider.AssignmentsBucket: + return "mesh-controller.assignments", nil + } + return "", fmt.Errorf("%s is not a bucket this provider keeps or reads", bucket) +} + +func (b busOf) Watch(ctx context.Context, bucket string, handle func(key string, value []byte)) error { + name, err := b.state(bucket) + if err != nil { + return err + } + stop, err := b.conn.StateWatch(Module, name, "", func(ch bus.StateChange) error { + if ch.Op == "delete" { + handle(ch.Key, nil) + } else { + handle(ch.Key, ch.Value) + } + return nil + }) + if err != nil { + return err + } + go func() { <-ctx.Done(); stop() }() + return nil +} + +func (b busOf) Put(_ context.Context, bucket, key string, value []byte) error { + name, err := b.state(bucket) + if err != nil { + return err + } + _, err = b.conn.StatePut(Module, name, key, json.RawMessage(value)) + return err +} + +func (b busOf) Delete(_ context.Context, bucket, key string) error { + name, err := b.state(bucket) + if err != nil { + return err + } + return b.conn.StateDelete(Module, name, key) +} + +// New is the helper answering node-tools's consumers on node. +func New(conn *bus.Conn, node string, every time.Duration, logf func(string, ...any)) *provider.Provider { + return provider.New(provider.Options{ + Node: node, + Module: Module, + Offers: Offers(), + Bus: busOf{conn: conn}, + Every: every, + Logf: logf, + }) +} + +// Run answers the consumers until ctx ends. The state is not reachable until the mesh has issued the runtime its +// membership and made the buckets, so a start that cannot watch is tried again, not given up. +func Run(ctx context.Context, conn *bus.Conn, node string, every time.Duration, logf func(string, ...any)) { + for delay := 2 * time.Second; ctx.Err() == nil; delay = min(delay*2, time.Minute) { + attempt, cancel := context.WithCancel(ctx) + err := New(conn, node, every, logf).Run(attempt) + cancel() + if ctx.Err() != nil { + return + } + logf("[mesh-tools] the provider helper stopped: %v; trying again in %s", err, delay) + select { + case <-ctx.Done(): + case <-time.After(delay): + } + } +} + +// Start runs the helper when this process is the machine's runtime (the node-tools credential, on a machine's +// name), with no condition on the environment, and returns what ends it. A runtime on another credential, a module's +// own in a test, is not the provider and answers nobody. +func Start(conn *bus.Conn, module, node string, every time.Duration, logf func(string, ...any)) (stop func()) { + if module != Module || node == "" { + return func() {} + } + ctx, cancel := context.WithCancel(context.Background()) + logf("[mesh-tools] answering %s's consumers of %s through the provider helper", Module, Provision) + go Run(ctx, conn, node, every, logf) + return cancel +} diff --git a/node-tools/internal/provide/provide_test.go b/node-tools/internal/provide/provide_test.go new file mode 100644 index 0000000..9f2b191 --- /dev/null +++ b/node-tools/internal/provide/provide_test.go @@ -0,0 +1,149 @@ +package provide + +import ( + "encoding/json" + "os" + "strings" + "testing" + "time" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// claude-code's facts.json template, as the catalogue keeps it (modules/claude-code/files/facts.json). +const factsTemplate = `{ + "node": "${machine:name}", + "console": "http://127.0.0.1:${bound:mcp-endpoint:port}/mcp" +} +` + +// factsToday is that file as the controller composes it today, from the port node-tools serves. +const factsToday = `{ + "node": "anchor", + "console": "http://127.0.0.1:4270/mcp" +} +` + +// fill is what the consumer's client does with the template: the machine's name, and each ${bound::} +// from the provider's entry. +func fill(t *testing.T, template string, entry map[string]any) string { + t.Helper() + values, _ := entry["values"].(map[string]any) + out := strings.ReplaceAll(template, "${machine:name}", "anchor") + for key, v := range values { + raw, _ := json.Marshal(v) + out = strings.ReplaceAll(out, "${bound:mcp-endpoint:"+key+"}", string(raw)) + } + return out +} + +func TestTheRuntimeAnswersAHandedConsumersMcpEndpointThroughTheHelper(t *testing.T) { + mesh := mt.New(t) + bound := mesh.Bucket(t, "node-tools_bound") + assignments := mesh.Bucket(t, "mesh-controller_assignments") + + mem := mt.MembershipOf("node-tools", "anchor", false, nil) + mem.State = []bus.StateIssued{ + {Name: "bound", Bucket: "node-tools_bound", Writes: true}, + {Name: "mesh-controller.assignments", Bucket: "mesh-controller_assignments"}, + } + mesh.Issue(t, mem) + + // claude-code, handed on this machine, is bound to node-tools here for mcp-endpoint (issue 788's plan entry). + plan := `{"node":"anchor","module":"claude-code","provisioning":{"provisions":[{"provision":"mcp-endpoint",` + + `"provider":"anchor","provider-module":"node-tools","at":"127.0.0.1","from":"anchor","as":"mesh_anchor_claude_code"}]}}` + if _, err := assignments.Put("anchor.claude-code", []byte(plan)); err != nil { + t.Fatal(err) + } + + conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"}) + if err != nil { + t.Fatal(err) + } + conn.Logf = func(string, ...any) {} + t.Cleanup(conn.Close) + conn.Follow("node-tools") + + said := &mt.Logs{} + t.Cleanup(Start(conn, "node-tools", "anchor", 20*time.Millisecond, said.Logf)) + + var entry map[string]any + mt.Until(t, func() error { + kept, err := bound.Get("anchor.claude-code.mcp-endpoint") + if err != nil { + return err + } + return json.Unmarshal(kept.Value(), &entry) + }) + values, _ := entry["values"].(map[string]any) + for key, want := range map[string]any{"port": float64(4270), "at": "127.0.0.1", "from": "anchor"} { + if values[key] != want { + t.Errorf("the entry's %s is %v, want %v (entry %v)", key, values[key], want, entry) + } + } + if entry["as"] != "mesh_anchor_claude_code" { + t.Errorf("the entry's as is %v", entry["as"]) + } + if c, has := entry["credential"]; has { + t.Errorf("an offer with no credential names one: %v", c) + } + if got := fill(t, factsTemplate, entry); got != factsToday { + t.Errorf("claude-code's facts.json fills to\n%s\nwant today's\n%s", got, factsToday) + } +} + +// The port the offer serves is the port the manifest serves and the console listens on. +func TestTheOfferServesThePortTheManifestServes(t *testing.T) { + raw, err := os.ReadFile("../../module.json") + if err != nil { + t.Fatal(err) + } + var m struct { + Serves map[string]map[string]any `json:"serves"` + Moved bool `json:"moved"` + Provides []struct { + Name string `json:"name"` + AnsweredBy string `json:"answered-by"` + } `json:"provides"` + Resources []struct { + ID string `json:"id"` + } `json:"resources"` + Build struct { + Artifacts []struct { + Name string `json:"name"` + Env map[string]string `json:"env"` + } `json:"artifacts"` + } `json:"build"` + } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + if got := m.Serves[Provision]["port"]; got != float64(ServedPort) { + t.Errorf("the manifest serves %v for %s, the offer %d", got, Provision, ServedPort) + } + if m.Moved { + t.Error("node-tools is marked moved: that also hands the runtime's bus account to the vault (issue 791 round 1)") + } + if len(m.Provides) != 1 || m.Provides[0].Name != Provision || m.Provides[0].AnsweredBy != "helper" { + t.Errorf("node-tools does not say its %s is answered by the helper: %+v", Provision, m.Provides) + } + for _, r := range m.Resources { + if r.ID == "consumers" { + t.Error("the manifest still carries a consumers file the helper does not need") + } + } + for _, a := range m.Build.Artifacts { + if a.Name == "files" || len(a.Env) > 0 { + t.Errorf("the build still carries %+v, which only a moved module needs", a) + } + } +} + +// Only the machine's runtime is the provider: a process on another credential, or with no machine, answers nobody. +func TestOnlyTheMachinesRuntimeStartsTheHelper(t *testing.T) { + said := &mt.Logs{} + for _, c := range []struct{ module, node string }{{"claude-code", "anchor"}, {"node-tools", ""}} { + Start(nil, c.module, c.node, time.Second, said.Logf)() + } +} diff --git a/node-tools/internal/runtime/provision.go b/node-tools/internal/runtime/provision.go new file mode 100644 index 0000000..c3c22b7 --- /dev/null +++ b/node-tools/internal/runtime/provision.go @@ -0,0 +1,216 @@ +package runtime + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "sync" + + "github.com/novox/mesh-tools/node-tools/internal/bus" +) + +// A module's code reads its values at runtime (novox/hq ADR 0349 clause 14 as ADR 0370 changes it, issue 685). +// +// A bundle asks `mesh/provision.get` and `mesh/provision.watch` with `{kind, key}`; the runtime asks its machine's +// provisioner client **as the module whose bundle asks** — the module this bus was made for, from the bundle it +// launched, never a name the bundle sent (ADR 0370 §4) — and hands the answer back. A watch hands the current value +// first and then each change: the client says which value moved, and the runtime asks again. A secret is refused +// here, with the reason, and never asked: the client serves no secret on the tool runner's word (§2), and a module +// under an account of its own asks the vault itself, which waits on issue 569. + +// The kinds a bundle may ask. +const ( + provisionConfig = "config" + provisionBound = "bound" + provisionSecret = "secret" +) + +// provisionAsked is what a bundle names. Anything else it sends, a module's name among it, is not read. +type provisionAsked struct { + Kind string `json:"kind"` + Key string `json:"key"` +} + +// ProvisionChange is what a watch hands a bundle: the value, or, when it is no longer answered, why. +type ProvisionChange struct { + Kind string `json:"kind"` + Key string `json:"key"` + Value json.RawMessage `json:"value,omitempty"` + // Refused is why the value is no longer answered: the module taken off the machine, the key no longer declared. + Refused string `json:"refused,omitempty"` +} + +func readProvisionAsked(verb string, params json.RawMessage) (provisionAsked, error) { + var asked provisionAsked + if err := json.Unmarshal(params, &asked); err != nil || asked.Kind == "" || asked.Key == "" { + return asked, fmt.Errorf("mesh/provision.%s names no value: {kind, key}", verb) + } + switch asked.Kind { + case provisionConfig, provisionBound: + return asked, nil + case provisionSecret: + return asked, fmt.Errorf("the secret %s is not asked for at runtime while the module's code runs in the "+ + "tool runner: the vault hands a secret only to a caller the bus proves, and the bus proves only the tool "+ + "runner here (novox/hq ADR 0370 §2, ADR 0349 §11); the module's secrets reach it through its files, "+ + "until it runs under a bus account of its own (issue 569)", asked.Key) + } + return asked, fmt.Errorf("mesh/provision.%s: %q is not a kind of value a module reads at runtime: config or bound", + verb, asked.Kind) +} + +// Provision answers a bundle's `get` of one of its module's values. +func (b *moduleBus) Provision(verb string, params json.RawMessage) (json.RawMessage, error) { + if verb != "get" { + return nil, fmt.Errorf("the runtime answers no mesh/provision.%s", verb) + } + asked, err := readProvisionAsked(verb, params) + if err != nil { + return nil, err + } + return b.all.conn.AskProvisioner(b.module, asked.Kind, asked.Key) +} + +// ProvisionWatch hands deliver the value now and each change after, until stopped. A value not answered now is +// refused now, with the client's reason. +func (b *moduleBus) ProvisionWatch(params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) { + asked, err := readProvisionAsked("watch", params) + if err != nil { + return nil, err + } + value, err := b.all.conn.AskProvisioner(b.module, asked.Kind, asked.Key) + if err != nil { + return nil, err + } + w := &provisionWatch{bus: b, asked: asked, deliver: deliver, last: value} + if err := w.hand(ProvisionChange{Kind: asked.Kind, Key: asked.Key, Value: value}); err != nil { + return nil, err + } + stop, err := b.all.provisioned.add(b.all, b.module, w) + if err != nil { + return nil, err + } + return stop, nil +} + +// provisionWatch is one watch a bundle holds: what it asked, and what it was handed last. +type provisionWatch struct { + bus *moduleBus + asked provisionAsked + deliver func(json.RawMessage) error + mu sync.Mutex // one change at a time, in order + last json.RawMessage + refused string + stopped bool +} + +func (w *provisionWatch) hand(change ProvisionChange) error { + raw, err := json.Marshal(change) + if err != nil { + return err + } + return w.deliver(raw) +} + +// again asks once more after the client said something moved, and hands the bundle what changed. +func (w *provisionWatch) again() { + w.mu.Lock() + defer w.mu.Unlock() + if w.stopped { + return + } + value, err := w.bus.all.conn.AskProvisioner(w.bus.module, w.asked.Kind, w.asked.Key) + change := ProvisionChange{Kind: w.asked.Kind, Key: w.asked.Key} + var refused *bus.ProvisionRefused + switch { + case errors.As(err, &refused): + if w.refused == refused.Reason { + return + } + change.Refused = refused.Reason + case err != nil: + // The client not reached: asked again on its next word, which it says whenever it comes up. + w.bus.all.logf("[mesh-tools] %s's watch of its %s %s could not ask again: %v", w.bus.module, w.asked.Kind, + w.asked.Key, err) + return + default: + if w.refused == "" && bytes.Equal(value, w.last) { + return + } + change.Value = value + } + if err := w.hand(change); err != nil { + w.bus.all.logf("[mesh-tools] %s did not take a change of its %s %s: %v", w.bus.module, w.asked.Kind, + w.asked.Key, err) + return + } + w.last, w.refused = value, change.Refused +} + +// provisionWatches is every watch on this runtime, by module, and the one subscription that hears the client. +type provisionWatches struct { + mu sync.Mutex + of map[string]map[*provisionWatch]bool + stop func() +} + +func (p *provisionWatches) add(all *consumers, module string, w *provisionWatch) (func(), error) { + p.mu.Lock() + defer p.mu.Unlock() + if p.stop == nil { + stop, err := all.conn.OnProvisionChanged(func(module string, changed bus.ProvisionChanged) { + for _, w := range p.told(module, changed) { + go w.again() + } + }) + if err != nil { + return nil, fmt.Errorf("the runtime cannot hear its machine's provisioner client: %w", err) + } + p.stop = stop + } + if p.of == nil { + p.of = map[string]map[*provisionWatch]bool{} + } + if p.of[module] == nil { + p.of[module] = map[*provisionWatch]bool{} + } + p.of[module][w] = true + return func() { + w.mu.Lock() + w.stopped = true + w.mu.Unlock() + p.mu.Lock() + defer p.mu.Unlock() + delete(p.of[module], w) + }, nil +} + +// told is the watches of module that what moved concerns. +func (p *provisionWatches) told(module string, changed bus.ProvisionChanged) []*provisionWatch { + p.mu.Lock() + defer p.mu.Unlock() + var out []*provisionWatch + for w := range p.of[module] { + if w.asked.Kind == changed.Kind && (changed.Key == "" || changed.Key == w.asked.Key) { + out = append(out, w) + } + } + return out +} + +func (p *provisionWatches) stopAll() { + p.mu.Lock() + defer p.mu.Unlock() + for _, ws := range p.of { + for w := range ws { + w.mu.Lock() + w.stopped = true + w.mu.Unlock() + } + } + p.of = nil + if p.stop != nil { + p.stop() + p.stop = nil + } +} diff --git a/node-tools/internal/runtime/provision_test.go b/node-tools/internal/runtime/provision_test.go new file mode 100644 index 0000000..f65b354 --- /dev/null +++ b/node-tools/internal/runtime/provision_test.go @@ -0,0 +1,203 @@ +package runtime + +import ( + "encoding/json" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// A module's code reads its values at runtime through moduleBus.Provision and ProvisionWatch (novox/hq ADR 0349 +// §14, ADR 0370, issue 685), against a real bus with a stand-in for the machine's provisioner client: the runtime +// asks as the module whose bundle asks, whatever the bundle names; a watch hands the current value, then each change +// the client says moved, and a refusal once; a secret is refused without asking. + +// standInClient answers prov.ask. from its values, keyed `..`, and keeps every ask. +type standInClient struct { + mu sync.Mutex + values map[string]string + asks []string + nc *nats.Conn +} + +func aStandInClient(t *testing.T, node string, values map[string]string) *standInClient { + t.Helper() + nc, err := nats.Connect(mt.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + c := &standInClient{values: values, nc: nc} + if _, err := nc.Subscribe("prov.ask."+node, func(m *nats.Msg) { + var a struct{ Module, Kind, Key string } + _ = json.Unmarshal(m.Data, &a) + c.mu.Lock() + c.asks = append(c.asks, string(m.Data)) + v, ok := c.values[a.Module+"."+a.Kind+"."+a.Key] + c.mu.Unlock() + if !ok { + _ = m.Respond([]byte(`{"error":"` + a.Module + ` declares no config value ` + a.Key + `"}`)) + return + } + _ = m.Respond([]byte(`{"value":` + v + `}`)) + }); err != nil { + t.Fatal(err) + } + _ = nc.Flush() + return c +} + +func (c *standInClient) set(key, value string) { + c.mu.Lock() + defer c.mu.Unlock() + if value == "" { + delete(c.values, key) + return + } + c.values[key] = value +} + +func (c *standInClient) asked() []string { + c.mu.Lock() + defer c.mu.Unlock() + return append([]string(nil), c.asks...) +} + +func (c *standInClient) say(t *testing.T, node, module, body string) { + t.Helper() + if err := c.nc.Publish("prov.changed."+node+"."+module, []byte(body)); err != nil { + t.Fatal(err) + } +} + +func TestABundleAsksForItsValuesAsItsOwnModule(t *testing.T) { + client := aStandInClient(t, "anchor", map[string]string{ + "claude-code.config.managed_settings": `{"spinnerTipsEnabled":false}`, + "dunst.config.managed_settings": `"dunst's"`, + }) + conn := connect(t, "node-tools", "anchor") + all := &consumers{conn: conn, logf: t.Logf, of: map[string]*moduleEvents{}} + b := all.forModule("claude-code").(*moduleBus) + + // Its own value, though the bundle names another module: the runtime names the module from the bundle. + got, err := b.Provision("get", json.RawMessage(`{"kind":"config","key":"managed_settings","module":"dunst"}`)) + if err != nil || string(got) != `{"spinnerTipsEnabled":false}` { + t.Fatalf("answered %s, %v", got, err) + } + if asks := client.asked(); len(asks) != 1 || asks[0] != `{"key":"managed_settings","kind":"config","module":"claude-code"}` { + t.Fatalf("the client was asked %v", asks) + } + // The client's refusal, in its words. + if _, err := b.Provision("get", json.RawMessage(`{"kind":"config","key":"nothing"}`)); err == nil || + !strings.Contains(err.Error(), "declares no config value nothing") { + t.Errorf("an undeclared key: %v", err) + } + // A secret, refused here and never asked; and what is not an ask. + before := len(client.asked()) + for params, want := range map[string]string{ + `{"kind":"secret","key":"gitea-read"}`: "the vault hands a secret only to a caller the bus proves", + `{"kind":"setting","key":"role"}`: "not a kind of value", + `{"key":"role"}`: "names no value", + } { + if _, err := b.Provision("get", json.RawMessage(params)); err == nil || !strings.Contains(err.Error(), want) { + t.Errorf("%s: %v, want %q", params, err, want) + } + } + if _, err := b.ProvisionWatch(json.RawMessage(`{"kind":"secret","key":"gitea-read"}`), func(json.RawMessage) error { + return nil + }); err == nil { + t.Error("a secret was watched") + } + if n := len(client.asked()); n != before { + t.Errorf("the client was asked for what the runtime refuses: %v", client.asked()[before:]) + } +} + +func TestABundlesWatchIsHandedTheValueAndEachChange(t *testing.T) { + client := aStandInClient(t, "anchor", map[string]string{"claude-code.config.managed_settings": `{"a":1}`}) + conn := connect(t, "node-tools", "anchor") + all := &consumers{conn: conn, logf: t.Logf, of: map[string]*moduleEvents{}} + t.Cleanup(all.stopAll) + b := all.forModule("claude-code").(*moduleBus) + + handed := make(chan ProvisionChange, 16) + stop, err := b.ProvisionWatch(json.RawMessage(`{"kind":"config","key":"managed_settings","module":"dunst"}`), + func(raw json.RawMessage) error { + var c ProvisionChange + if err := json.Unmarshal(raw, &c); err != nil { + t.Errorf("handed %s", raw) + } + handed <- c + return nil + }) + if err != nil { + t.Fatal(err) + } + next := func(what string) ProvisionChange { + t.Helper() + select { + case c := <-handed: + return c + case <-time.After(5 * time.Second): + t.Fatalf("nothing handed: %s", what) + } + return ProvisionChange{} + } + nothing := func(what string) { + t.Helper() + select { + case c := <-handed: + t.Errorf("%s: handed %+v", what, c) + case <-time.After(300 * time.Millisecond): + } + } + if c := next("the current value"); string(c.Value) != `{"a":1}` || c.Kind != "config" || c.Key != "managed_settings" { + t.Fatalf("first handed %+v", c) + } + + // A change, said by the client: asked again, handed. + client.set("claude-code.config.managed_settings", `{"a":2}`) + client.say(t, "anchor", "claude-code", `{"kind":"config","key":"managed_settings"}`) + if c := next("the change"); string(c.Value) != `{"a":2}` { + t.Fatalf("after the change, handed %+v", c) + } + // Every key of the kind said to have moved, the value the same: nothing handed. Another module's or another + // key's word: not even asked. + client.say(t, "anchor", "claude-code", `{"kind":"config"}`) + nothing("the same value again") + before := len(client.asked()) + client.say(t, "anchor", "dunst", `{"kind":"config","key":"managed_settings"}`) + client.say(t, "anchor", "claude-code", `{"kind":"config","key":"role"}`) + nothing("another's word") + if n := len(client.asked()); n != before { + t.Errorf("asked again for another's word: %v", client.asked()[before:]) + } + // No longer answered: handed the refusal, once. + client.set("claude-code.config.managed_settings", "") + client.say(t, "anchor", "claude-code", `{"kind":"config"}`) + if c := next("the refusal"); c.Refused == "" || len(c.Value) != 0 { + t.Fatalf("handed %+v, want the refusal", c) + } + client.say(t, "anchor", "claude-code", `{"kind":"config"}`) + nothing("the same refusal again") + // Answered again, and then stopped: nothing after. + client.set("claude-code.config.managed_settings", `{"a":3}`) + client.say(t, "anchor", "claude-code", `{"kind":"config"}`) + if c := next("answered again"); string(c.Value) != `{"a":3}` { + t.Fatalf("handed %+v", c) + } + stop() + client.set("claude-code.config.managed_settings", `{"a":4}`) + client.say(t, "anchor", "claude-code", `{"kind":"config"}`) + nothing("a stopped watch") + for _, a := range client.asked() { + if !strings.Contains(a, `"module":"claude-code"`) { + t.Errorf("asked as another module: %s", a) + } + } +} diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index a98e425..7bd9a8d 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -458,6 +458,8 @@ type consumers struct { of map[string]*moduleEvents // seatBound is each module's seat traffic bound once: a worker taken, a proof subject answered. seatBound map[string]func() + // provisioned is every watch a module's code holds of its values at runtime (provision.go). + provisioned provisionWatches } type moduleEvents struct { @@ -479,6 +481,7 @@ func (c *consumers) stopAll() { stop() } c.seatBound = map[string]func(){} + c.provisioned.stopAll() } // forModule is the bus one launched bundle of a module reaches the mesh through. diff --git a/node-tools/module.json b/node-tools/module.json index 8c02041..6eb1fce 100644 --- a/node-tools/module.json +++ b/node-tools/module.json @@ -15,7 +15,8 @@ "provides": [ { "name": "mcp-endpoint", - "scope": "node" + "scope": "node", + "answered-by": "helper" } ], "serves": { @@ -59,5 +60,11 @@ "binary": "node-tools" } ] - } + }, + "state": [ + "bound" + ], + "reads": [ + "mesh-controller.assignments" + ] }