Tool runner: a bundle's over-long line ends the bundle and is said; an unanswered call and a refused answer are logged (issue 1966)

The reader of a bundle's stdout stopped at a line over 16 MiB without a word and the bundle's next write
blocked for good: every call of the module then timed out, unheard of. Now the line is said, the bundle ended
and started again by the exit path. A call the bundle does not answer in time is logged with tool and caller,
and a refused publish of an answer (to a caller's inbox) is said once per subject per minute.

Quick fix: only the bundle that met it (mesh-hand-acts' answer is cut by its own module). Thorough fix, chosen:
the runner guards every bundle, so the kind (a silent freeze of a module runtime) cannot recur for any module.

Claude-Session: https://claude.ai/code/session_01NcbgfFfQ4Z9UE5jK32PmE5

node-tools/internal/bus/bus.go

@@ -195,6 +195,8 @@ type Conn struct {
 	calls callers
 	// waits are the waits in flight by subject, each told when the bus refuses its subscription (wait.go).
 	waits map[string]map[chan string]bool
+	// inboxSaid is when a refused answer publish was last said, by subject (answerRefused).
+	inboxSaid map[string]time.Time
 }
 
 // answered is one subject the runtime answers on: what to subscribe again with, and how often it was
@@ -232,6 +234,9 @@ func (c *Conn) refused(err error) {
 	if c.publishRefused(err) {
 		return
 	}
+	if c.answerRefused(err) {
+		return
+	}
 	if subject := subjectRefused(err); subject != "" && c.waitRefused(subject) {
 		return
 	}
@@ -1215,3 +1220,28 @@ func keyOf(subject string) string {
 	}
 	return event
 }
+
+// answerRefused says a refused publish of an answer (to a caller's inbox: a permission, or the single-use
+// allowance of a reply expired) once per subject per minute, never silently (issue 1966: a caller waited out its
+// time for an answer the bus had refused). It reports whether the refusal was one.
+func (c *Conn) answerRefused(err error) bool {
+	m := refusedPublish.FindStringSubmatch(err.Error())
+	if len(m) < 2 || !strings.HasPrefix(strings.Trim(m[1], `"`), "_INBOX.") {
+		return false
+	}
+	subject := strings.Trim(m[1], `"`)
+	c.mu.Lock()
+	if c.inboxSaid == nil {
+		c.inboxSaid = map[string]time.Time{}
+	}
+	last, seen := c.inboxSaid[subject]
+	say := !seen || time.Since(last) >= time.Minute
+	if say {
+		c.inboxSaid[subject] = time.Now()
+	}
+	c.mu.Unlock()
+	if say && c.Logf != nil {
+		c.Logf("[mesh-tools] the bus refused this account an answer to %s (%v); the caller gets none", subject, err)
+	}
+	return true
+}

node-tools/internal/bus/refused_test.go

@@ -84,3 +84,28 @@ func TestARefusedSubscriptionIsAskedForAgain(t *testing.T) {
 		t.Error("a stopped subject is still tracked")
 	}
 }
+
+// Issue 1966: a refused publish of an answer (to a caller's inbox) is said, once per subject per minute, never dropped
+// in silence while the caller waits out its time.
+func TestARefusedAnswerIsSaidOncePerMinute(t *testing.T) {
+	c := &Conn{}
+	var said []string
+	c.Logf = func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }
+	refuse := func(subject string) bool {
+		return c.answerRefused(fmt.Errorf("%w: Permissions Violation for Publish to %q", nats.ErrPermissionViolation, subject))
+	}
+	if !refuse("_INBOX.user.abc") || !refuse("_INBOX.user.abc") || !refuse("_INBOX.user.def") {
+		t.Fatal("a refused answer was not taken for one")
+	}
+	if len(said) != 2 {
+		t.Fatalf("want one line per subject, got %v", said)
+	}
+	c.inboxSaid["_INBOX.user.abc"] = time.Now().Add(-2 * time.Minute)
+	refuse("_INBOX.user.abc")
+	if len(said) != 3 || !strings.Contains(said[2], "_INBOX.user.abc") {
+		t.Fatalf("a minute later it is said again, got %v", said)
+	}
+	if refuse("mesh.mod.x.tool.y") {
+		t.Fatal("a refused publish that is no answer was taken for one")
+	}
+}

node-tools/internal/launch/launch.go

@@ -211,6 +211,26 @@ type child struct {
 	watches []func()
 }
 
+// maxLine is the longest line the runner reads from a bundle; a variable so a test need not write megabytes.
+var maxLine = 16 << 20
+
+// errNoAnswer is a call the bundle did not answer within its bound.
+var errNoAnswer = errors.New("the bundle did not answer in time")
+
+// readCapped reads one line of at most max bytes. tooLong says it was longer; the rest is not read.
+func readCapped(r *bufio.Reader, max int) (line []byte, tooLong bool, err error) {
+	for {
+		part, isPrefix, e := r.ReadLine()
+		line = append(line, part...)
+		if len(line) > max {
+			return nil, true, nil
+		}
+		if e != nil || !isPrefix {
+			return line, false, e
+		}
+	}
+}
+
 func (c *child) write(m any) error {
 	b, err := wire.Marshal(m)
 	if err != nil {
@@ -324,10 +344,22 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
 		}()
 		go func() {
 			defer close(c.handled)
-			scan := bufio.NewScanner(stdout)
-			scan.Buffer(make([]byte, 64*1024), 16<<20)
-			for scan.Scan() {
-				line := strings.TrimSpace(scan.Text())
+			rd := bufio.NewReaderSize(stdout, 64*1024)
+			for {
+				raw, tooLong, rerr := readCapped(rd, maxLine)
+				if tooLong {
+					// Never left frozen (issue 1966): a reader that stops while the bundle lives leaves the bundle's
+					// next write blocked for good. Said, and the bundle ended; the exit path below fails what waits
+					// on it and starts it again.
+					logf("[mesh-tools] %s's bundle wrote a line over %d bytes (a tool's answer the bus cannot carry); ending the bundle",
+						module, maxLine)
+					_ = cmd.Process.Kill()
+					break
+				}
+				if rerr != nil && len(raw) == 0 {
+					break
+				}
+				line := strings.TrimSpace(string(raw))
 				if line == "" {
 					continue
 				}
@@ -697,6 +729,10 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
 				defer end()
 				res, err := asking("tools/call", callParams(full, caller, call, args), CallTimeout)
 				if err != nil {
+					if errors.Is(err, errNoAnswer) {
+						// Its caller is gone by now and nobody else would hear of it (issue 1966).
+						logf("[mesh-tools] %s.%s called by %q: %v", module, full, caller, err)
+					}
 					return nil, err
 				}
 				var called struct {
@@ -794,7 +830,7 @@ func (c *child) ask(module, method string, params any, timeout time.Duration) (j
 		c.mu.Lock()
 		delete(c.pending, id)
 		c.mu.Unlock()
-		return nil, fmt.Errorf("%s's bundle did not answer %s in %ds", module, method, int(timeout/time.Second))
+		return nil, fmt.Errorf("%s's bundle did not answer %s in %ds: %w", module, method, int(timeout/time.Second), errNoAnswer)
 	}
 }
 

node-tools/internal/launch/state_failed_test.go

@@ -73,3 +73,67 @@ func TestAFailedStateOperationIsSaidInTheRuntimesLog(t *testing.T) {
 		}
 	}
 }
+
+// A bundle that lists one tool, "t", and answers a call to it as its MODE says: "long" with a line over the runner's cap,
+// "silent" with nothing at all.
+const callingBundle = `#!/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":[{"name":"t","description":"d"}]}}\n' "$id" ;;
+    *'"method":"tools/call"'*) [ "$MODE" = long ] && printf '%4096s\n' x ;;
+  esac
+done
+`
+
+// Issue 1966: a bundle's line over the runner's cap, and a call the bundle never answers, are said in the runtime's log;
+// the over-long line ends the bundle and fails the call at once, never leaving the runner reading nothing while the
+// bundle's next write blocks.
+func TestAnOverlongLineAndAnUnansweredCallAreSaid(t *testing.T) {
+	oldMax, oldCall := maxLine, CallTimeout
+	maxLine, CallTimeout = 1024, 300*time.Millisecond
+	t.Cleanup(func() { maxLine, CallTimeout = oldMax, oldCall })
+	for _, row := range []struct{ mode, said string }{
+		{"long", "bundle wrote a line over 1024 bytes"},
+		{"silent", ""},
+	} {
+		t.Run(row.mode, func(t *testing.T) {
+			entry := filepath.Join(t.TempDir(), "bundle")
+			if err := os.WriteFile(entry, []byte(callingBundle), 0o755); err != nil {
+				t.Fatal(err)
+			}
+			said := make(chan string, 32)
+			logf := func(f string, a ...any) { said <- fmt.Sprintf(f, a...) }
+			l, err := Start("caller", entry, append(os.Environ(), "MODE="+row.mode), &refusingState{}, logf)
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(l.Stop)
+			begun := time.Now()
+			_, err = l.Registrations[0].Tools[0].Run("caller-1", nil)
+			if err == nil {
+				t.Fatal("the call was answered")
+			}
+			if took := time.Since(begun); took > 5*time.Second {
+				t.Fatalf("the call hung for %s", took)
+			}
+			var heard []string
+			timeout := time.After(5 * time.Second)
+			for {
+				select {
+				case line := <-said:
+					heard = append(heard, line)
+					if row.mode == "long" && strings.Contains(line, row.said) && strings.Contains(line, "ending the bundle") {
+						return
+					}
+					if row.mode == "silent" && strings.Contains(line, "caller.t called by \"caller-1\"") && strings.Contains(line, "did not answer tools/call") {
+						return
+					}
+				case <-timeout:
+					t.Fatalf("not said; the log: %v", heard)
+				}
+			}
+		})
+	}
+}