README.md

@@ -54,6 +54,36 @@ 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.
 
+#### Taking an event — the one rule
+
+A module's handler (`on(pattern, handler)` here, `stdio.Subscribe` in Go, the same in every language)
+**takes an event by returning, and asks for it again by throwing** (Go: returning an error). Nothing
+else counts: what a handler returns is not read.
+
+On the wire, the runtime hands the bundle `mesh/event` with the envelope and acknowledges the event on
+the bus when the answer is **anything but a JSON-RPC error** — `{}` (what the SDKs send), `null`, or no
+`result` at all. An `error` answer, no answer within the runtime's event timeout (two minutes), or the
+bundle exiting leaves it unacknowledged: it is offered again after five seconds, up to the module
+consumer's maximum deliveries, after which the bus gives it up and the controller raises a condition
+(`max-deliveries`). The runtime's line for an event not taken quotes a handler's error as the
+handler's own words, so it is never mistaken for the runtime's.
+
+So, for a handler:
+
+- **Throw only when the work was not done** — the service refused, was unreachable, the store failed.
+  Never for something that went right but answered oddly: an API that answers an empty body is a
+  success, and reading it as JSON is a throw that has every event offered five times, its work done
+  five times, and a false alarm raised (novox/hq issue 276).
+- **Delivery is at-least-once**, so the work must be safe to do twice, or the handler skips an event
+  whose `x-event-id` (`event.id`) it already did — remembering it only once the work succeeded, so one
+  that failed is done when it is offered again.
+- An event no handler matches is answered at once and taken.
+- A subscribe the runtime refused leaves no handler behind, so asking again until it is taken hands
+  each event to the handler once.
+
+Checked by the SDKs' stdio tests (TypeScript `test/stdio.test.ts`, Go `go/stdio_test.go`) and the
+runtime's (`node-tools/internal/launch`, mesh-tools).
+
 ### `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

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, 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.
+`stdio.Emit(type, body)` publishes an event as the module, `stdio.Subscribe(pattern, handler)` receives the module's events (taken when the handler returns nil, offered again when it returns an error — the one rule is in the repository README, "Taking an event") 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/stdio.go

@@ -60,11 +60,14 @@ type channel struct {
 	next      int
 	waiting   map[string]chan message
 	listeners []listener
+	// subscribed numbers the listeners, so one whose subscribe failed is the one removed.
+	subscribed int
 	// 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 {
+	n       int
 	pattern string
 	handle  func(Envelope) error
 }
@@ -108,7 +111,7 @@ func (c *channel) askAs(id, method string, params any, timeout time.Duration) (j
 	select {
 	case m := <-answer:
 		if m.Error != nil {
-			return nil, errors.New(m.Error.Message)
+			return nil, &refused{m.Error.Message}
 		}
 		return m.Result, nil
 	case <-time.After(timeout):
@@ -119,6 +122,11 @@ func (c *channel) askAs(id, method string, params any, timeout time.Duration) (j
 	}
 }
 
+// refused is the runtime's error answer, as opposed to no answer: what it refused, it did not do.
+type refused struct{ message string }
+
+func (r *refused) Error() string { return r.message }
+
 var (
 	current   *channel
 	currentMu sync.Mutex
@@ -287,18 +295,43 @@ func live() (*channel, error) {
 }
 
 // Subscribe asks the runtime for this module's events whose type matches the pattern (`*` one
-// segment, `#` any), and hands each to handle (ADR 0198). The runtime acknowledges an event on the bus
-// only after every matching handler returned nil; an error has it delivered again. Call it from a
-// goroutine started before Serve, or from a tool, once Serve runs.
+// segment, `#` any), and hands each to handle (ADR 0198).
+//
+// **Taking an event** (see the repository's README): handle returning nil takes the event — the
+// runtime acknowledges it once every matching handler did; an error is the one way to have it offered
+// again, after a few seconds and a bounded number of times, after which the bus gives up and the
+// controller raises a condition. So return an error only when the work was not done, never for an
+// answer that went right but read oddly; and since delivery is at-least-once, make the work safe to
+// do twice, or skip an event whose x-event-id was already done.
+//
+// A subscribe the runtime refused leaves no handler behind, so asking again — as a module does until
+// the runtime takes it — does not hand one event to the same handler twice. (One the runtime did not
+// answer in time is kept, as the runtime may have bound it.) Call it from a goroutine
+// started before Serve, or from a tool, once Serve runs.
 func Subscribe(pattern string, handle func(Envelope) error) error {
 	c, err := live()
 	if err != nil {
 		return err
 	}
 	c.mu.Lock()
-	c.listeners = append(c.listeners, listener{pattern: pattern, handle: handle})
+	c.subscribed++
+	n := c.subscribed
+	c.listeners = append(c.listeners, listener{n: n, pattern: pattern, handle: handle})
 	c.mu.Unlock()
 	_, err = c.ask("mesh/subscribe", map[string]any{"pattern": pattern}, 30*time.Second)
+	// Removed only when the runtime answered that it refused: one that did not answer in time may
+	// have bound the consumer, and a handler removed then would have its events taken unhandled.
+	var no *refused
+	if errors.As(err, &no) {
+		c.mu.Lock()
+		for i, l := range c.listeners {
+			if l.n == n {
+				c.listeners = append(c.listeners[:i:i], c.listeners[i+1:]...)
+				break
+			}
+		}
+		c.mu.Unlock()
+	}
 	return err
 }
 

go/stdio_test.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"io"
+	"sync"
 	"testing"
 	"time"
 )
@@ -139,3 +140,87 @@ func TestABundleSubscribesAndAsksThroughTheRuntime(t *testing.T) {
 		t.Error("topic matching differs from the TypeScript SDK's")
 	}
 }
+
+// A subscribe the runtime refused leaves no handler behind: asked again until the runtime takes it, a
+// module's handler is handed each event once, not once per attempt (novox/hq issue 276).
+func TestARefusedSubscribeLeavesNoHandlerBehind(t *testing.T) {
+	t.Setenv(ServedAs, "plex")
+	inR, inW := io.Pipe()
+	outR, outW := io.Pipe()
+	var mu sync.Mutex
+	handled := 0
+	subscribed := make(chan error, 2)
+	go func() {
+		_ = ServeOn("", []Tool{{Name: "start", Run: func(map[string]any) (any, error) {
+			go func() {
+				for i := 0; i < 2; i++ {
+					err := Subscribe("*.download.completed", func(Envelope) error {
+						mu.Lock()
+						handled++
+						mu.Unlock()
+						return nil
+					})
+					subscribed <- err
+					if err == nil {
+						return
+					}
+				}
+			}()
+			return "started", nil
+		}}}, 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"}})
+	answered := 0
+	for answered < 2 {
+		m := next()
+		switch m["method"] {
+		case "mesh/subscribe":
+			if answered == 0 {
+				send(map[string]any{"jsonrpc": "2.0", "id": m["id"], "error": map[string]any{"code": -32000, "message": "not yet"}})
+			} else {
+				send(map[string]any{"jsonrpc": "2.0", "id": m["id"], "result": map[string]any{}})
+			}
+			answered++
+		}
+	}
+	if err := <-subscribed; err == nil || err.Error() != "not yet" {
+		t.Fatalf("the refused subscribe said %v", err)
+	}
+	if err := <-subscribed; err != nil {
+		t.Fatalf("the second subscribe said %v", err)
+	}
+	send(map[string]any{"jsonrpc": "2.0", "id": 9, "method": "mesh/event", "params": map[string]any{"envelope": map[string]any{"key": "radarr.download.completed", "node": "anchor", "body": map[string]any{}}}})
+	for {
+		if r := next(); r["id"] == float64(9) {
+			if r["error"] != nil {
+				t.Fatalf("the event was answered %v", r)
+			}
+			break
+		}
+	}
+	_ = inW.Close()
+	mu.Lock()
+	defer mu.Unlock()
+	if handled != 1 {
+		t.Fatalf("one event was handled %d times", handled)
+	}
+}

package-lock.json

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

package.json

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

src/events/index.ts

@@ -65,6 +65,10 @@ export async function emit<T>(type: string, body: T, opts: EmitOptions = {}): Pr
  * React to events whose type matches a topic pattern (`*` one segment, `#` any). The audit logger
  * is just `on("#", …)`. The handler receives the reconstructed event — its metadata read back from
  * the headers, its body the domain payload.
+ *
+ * **The handler takes the event by returning and asks for it again by throwing** — nothing else
+ * counts (README, "Taking an event"). Throw only when the work was not done; delivery is
+ * at-least-once, so make the work safe to repeat or skip an `id` already done.
  */
 export async function on<T>(pattern: string, handler: (event: Event<T>) => Promise<void>): Promise<() => void> {
   return broker().subscribe<T>(pattern, async (envelope) => {

src/stdio/index.ts

@@ -106,11 +106,19 @@ function startSession(): Session {
     subscribe: (async <T>(pattern: string, handler: (env: Envelope<T>) => Promise<void>) => {
       const listener = { pattern, handler: handler as (env: Envelope<unknown>) => Promise<void> };
       listeners.push(listener);
-      await ask("mesh/subscribe", { pattern });
-      return () => {
+      const remove = (): void => {
         const at = listeners.indexOf(listener);
         if (at >= 0) listeners.splice(at, 1);
       };
+      try {
+        await ask("mesh/subscribe", { pattern });
+      } catch (err) {
+        // Refused: nothing is bound, and a handler left behind would be handed each event twice once
+        // the module asks again (novox/hq issue 276).
+        remove();
+        throw err;
+      }
+      return remove;
     }) as Broker["subscribe"],
     close: async () => {},
   };

test/fixtures/stdio-resubscribe.mjs

@@ -0,0 +1,18 @@
+// A launched bundle that asks for its events until the runtime takes the subscription — as a module
+// whose first ask may come too early does — and counts how often its one handler is handed an event.
+import { on } from "../../dist/events/index.js";
+import { serveRegisteredOverStdio } from "../../dist/stdio/index.js";
+
+let handled = 0;
+for (;;) {
+  try {
+    await on("radarr.download.completed", async () => {
+      handled++;
+      process.stderr.write(`handled ${handled}\n`);
+    });
+    break;
+  } catch (err) {
+    process.stderr.write(`refused: ${err.message}\n`);
+  }
+}
+await serveRegisteredOverStdio();

test/stdio.test.ts

@@ -160,6 +160,40 @@ test("a launched bundle subscribes as it is imported, handles events the runtime
   }
 });
 
+test("a subscribe the runtime refused leaves no handler behind, so one event is handled once (novox/hq issue 276)", async () => {
+  const child = spawn(process.execPath, [fileURLToPath(new URL("./fixtures/stdio-resubscribe.mjs", import.meta.url))],
+    { stdio: ["pipe", "pipe", "pipe"], env: { ...process.env, MESH_SERVED_MODULE: "plex" } });
+  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");
+  try {
+    const first = await until((m) => m.method === "mesh/subscribe");
+    send({ jsonrpc: "2.0", id: first.id, error: { code: -32000, message: "not yet" } });
+    const second = await until((m) => m.method === "mesh/subscribe" && m.id !== first.id);
+    send({ jsonrpc: "2.0", id: second.id, result: {} });
+    send({ jsonrpc: "2.0", id: 20, method: "mesh/event", params: { envelope: { key: "radarr.download.completed", node: "anchor", body: {}, headers: { "x-event-id": "e20" } } } });
+    assert.deepEqual((await until((m) => m.id === 20)).result, {});
+    assert.match(said, /refused: not yet/);
+    assert.equal(said.match(/handled/g)?.length, 1, `one event was handled more than once:\n${said}`);
+  } finally {
+    child.kill();
+  }
+});
+
 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" } });