Ask a declared tool check on the NATS link the node-engine holds
mesh/merge-gate pass: builds mesh-host → ace, g14, novox, shanks; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery superseded: a newer head of the same pull request

The queue asks a tool through a ToolBus assertion that only OverNATS satisfied; the
link the engine actually holds (natsLink) had no Ask, so every tool check failed as
"no link to the bus is open" while the link was up (hq issue 331). Give natsLink Ask,
assert at build time every optional interface the queue expects of it, and tell a link
that cannot ask apart from no link.
This commit is contained in:
jochen
2026-10-08 16:44:01 +02:00
parent 6400e92d19
commit 809ab4cf29
4 changed files with 103 additions and 2 deletions
+40
View File
@@ -0,0 +1,40 @@
package link
import (
"context"
"strings"
"testing"
"time"
"github.com/nats-io/nats.go"
)
// A declared tool check asked over a real bus, through the queue, on the link the node-engine holds:
// the tool's answer is read, and a tool nothing serves is said as such (novox/hq ADR 0240, to-be 48 §3).
func TestNatsAToolCheckIsAskedOnTheLink(t *testing.T) {
conn, js := aBus(t)
const node = "toolcheck"
served, err := conn.Subscribe(ToolSubject("sensors", "sensors_health", node), func(m *nats.Msg) {
_ = m.Respond([]byte(`{"result":{"healthy":true}}`))
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = served.Unsubscribe() })
if err := conn.Flush(); err != nil {
t.Fatal(err)
}
q := &Queue{Membership: Membership{Node: node}}
q.attach(t.Context(), &natsLink{conn: conn, js: js, node: node})
ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
defer cancel()
healthy, why, err := q.AskTool(ctx, "sensors", "sensors_health")
if err != nil || !healthy {
t.Fatalf("a served healthy tool read as %v %q (%v)", healthy, why, err)
}
_, _, err = q.AskTool(ctx, "sensors", "nothing_serves_this")
if err == nil || !strings.Contains(err.Error(), "nothing on this machine answers") {
t.Fatalf("a tool nothing serves read as %v", err)
}
}
+31
View File
@@ -3,6 +3,7 @@ package link
import (
"context"
"encoding/json"
"strings"
"testing"
"time"
)
@@ -70,3 +71,33 @@ func TestAToolsAnswerIsHealthyOnlyWhenItSaysSo(t *testing.T) {
t.Errorf("asked on %s", ToolSubject("keycloak", "keycloak_admin_health", "anchor"))
}
}
// A declared tool check is asked on the link the node-engine actually holds — the NATS link — and not
// refused as if none were open (2026-10-08: every tool check read "no link to the bus is open" while
// the link was up, because the NATS link could not Ask). With no link open, and with a link that cannot
// ask, the check fails and says which.
func TestAToolCheckIsAskedOnTheLinkTheEngineHolds(t *testing.T) {
q := &Queue{Membership: Membership{Node: "anchor"}}
if _, _, err := q.AskTool(t.Context(), "sensors", "sensors_health"); err == nil || err.Error() != "no link to the bus is open" {
t.Fatalf("with no link open: %v", err)
}
quiet := newQuietLink()
q.attach(t.Context(), quiet)
if _, _, err := q.AskTool(t.Context(), "sensors", "sensors_health"); err == nil || err.Error() != "the link to the bus open now cannot ask a tool" {
t.Fatalf("on a link that cannot ask: %v", err)
}
q.detach(quiet)
// The link Open returns, with no connection under it: the question must reach its Ask (and fail
// there), never stop at the queue's assertion.
held := &natsLink{node: "anchor"}
q.attach(t.Context(), held)
_, _, err := q.AskTool(t.Context(), "sensors", "sensors_health")
if err == nil {
t.Fatal("asked on no connection and was answered")
}
if strings.Contains(err.Error(), "no link") || strings.Contains(err.Error(), "cannot ask a tool") {
t.Fatalf("the NATS link was not asked: %v", err)
}
}
+27
View File
@@ -220,6 +220,33 @@ func (l *natsLink) Health(ctx context.Context, node string, body []byte) error {
return OverNATS{Conn: l.conn, JS: l.js}.Health(ctx, node, body)
}
// Ask asks a module's tool for a declared tool check (novox/hq ADR 0240, to-be 48 §3). The link the
// node-engine holds is this one, not OverNATS: without it here the queue's ToolBus assertion failed on
// every live link and each tool check read "no link to the bus is open" while the link was up.
func (l *natsLink) Ask(ctx context.Context, subject string, body []byte) ([]byte, error) {
reply, err := OverNATS{Conn: l.conn, JS: l.js}.Ask(ctx, subject, body)
switch {
case errors.Is(err, nats.ErrNoResponders):
return nil, fmt.Errorf("nothing on this machine answers %s", subject)
case err != nil:
if refused := l.refusal(subject); refused != "" {
return nil, fmt.Errorf("asking %s: %v%s", subject, err, refused)
}
return nil, fmt.Errorf("asking %s: %w", subject, err)
}
return reply, nil
}
// What the queue asks of a live link beside Bus, each by a type assertion that fails quietly: said here
// so a link that drops one does not build.
var (
_ Bus = (*natsLink)(nil)
_ HealthBus = (*natsLink)(nil)
_ ToolBus = (*natsLink)(nil)
_ Asker = (*natsLink)(nil)
_ Asked = (*natsLink)(nil)
)
// natsDeclaration is one declaration off the NODES stream.
type natsDeclaration struct{ msg *nats.Msg }
+5 -2
View File
@@ -530,10 +530,13 @@ func (q *Queue) AskTool(ctx context.Context, module, tool string) (bool, string,
q.mu.Lock()
bus := q.bus
q.mu.Unlock()
asker, ok := bus.(ToolBus)
if bus == nil || !ok {
if bus == nil {
return false, "", errors.New("no link to the bus is open")
}
asker, ok := bus.(ToolBus)
if !ok {
return false, "", errors.New("the link to the bus open now cannot ask a tool")
}
reply, err := asker.Ask(ctx, ToolSubject(module, tool, q.Membership.Node), []byte("{}"))
if err != nil {
return false, "", err