Tool runner: the cap's reason fails the call; the test proves the ending; the refused-answer limit keys on the caller's prefix and forgets old entries (issue 1966, review round 1)

Claude-Session: https://claude.ai/code/session_01NcbgfFfQ4Z9UE5jK32PmE5
2026-11-09 13:59:13 +01:00
parent f3a9cc276e
commit 312e29478e
4 changed files with 73 additions and 27 deletions
@@ -1229,15 +1229,25 @@ func (c *Conn) answerRefused(err error) bool {
 	if len(m) < 2 || !strings.HasPrefix(strings.Trim(m[1], `"`), "_INBOX.") {
 		return false
 	}
+	// Keyed without the reply's last token (a new one per request), so a caller's burst is one line.
 	subject := strings.Trim(m[1], `"`)
+	if dot := strings.LastIndex(subject, "."); dot > len("_INBOX.") {
+		subject = subject[:dot]
+	}
 	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
+	now := time.Now()
+	for k, at := range c.inboxSaid {
+		if now.Sub(at) >= time.Minute {
+			delete(c.inboxSaid, k)
+		}
+	}
+	_, seen := c.inboxSaid[subject]
+	say := !seen
 	if say {
-		c.inboxSaid[subject] = time.Now()
+		c.inboxSaid[subject] = now
 	}
 	c.mu.Unlock()
 	if say && c.Logf != nil {
@@ -94,17 +94,22 @@ func TestARefusedAnswerIsSaidOncePerMinute(t *testing.T) {
 	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") {
+	// A reply subject ends in a token new to every request: a burst under one prefix is one line.
+	if !refuse("_INBOX.user.abc.t1") || !refuse("_INBOX.user.abc.t2") || !refuse("_INBOX.user.def.t1") {
 		t.Fatal("a refused answer was not taken for one")
 	}
 	if len(said) != 2 {
-		t.Fatalf("want one line per subject, got %v", said)
+		t.Fatalf("want one line per caller, got %v", said)
 	}
 	c.inboxSaid["_INBOX.user.abc"] = time.Now().Add(-2 * time.Minute)
-	refuse("_INBOX.user.abc")
+	c.inboxSaid["_INBOX.user.old"] = time.Now().Add(-2 * time.Minute)
+	refuse("_INBOX.user.abc.t3")
 	if len(said) != 3 || !strings.Contains(said[2], "_INBOX.user.abc") {
 		t.Fatalf("a minute later it is said again, got %v", said)
 	}
+	if _, kept := c.inboxSaid["_INBOX.user.old"]; kept || len(c.inboxSaid) != 2 {
+		t.Fatalf("entries older than a minute are dropped on write, kept %v", c.inboxSaid)
+	}
 	if refuse("mesh.mod.x.tool.y") {
 		t.Fatal("a refused publish that is no answer was taken for one")
 	}
@@ -206,6 +206,8 @@ type child struct {
 	// restart decided — which Stop waits for (novox/hq issue 476).
 	handled chan struct{}
 	why     error
+	// endedFor is why the runner itself ended the bundle (a line over the cap); its calls fail with it.
+	endedFor string
 	// watches are the state watches this child asked for, stopped when it exits (ADR 0201): a child
 	// that starts again watches again, as its code runs again.
 	watches []func()
@@ -351,8 +353,11 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
 					// 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)
+					reason := fmt.Sprintf("%s's bundle wrote a line over %d bytes (a tool's answer the bus cannot carry)", module, maxLine)
+					logf("[mesh-tools] %s; ending the bundle", reason)
+					c.mu.Lock()
+					c.endedFor = reason
+					c.mu.Unlock()
 					_ = cmd.Process.Kill()
 					break
 				}
@@ -561,6 +566,9 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
 			}
 			saidMu.Unlock()
 			c.mu.Lock()
+			if c.endedFor != "" {
+				why = c.endedFor + "; the bundle was ended (" + code + ")"
+			}
 			c.why = errors.New(why)
 			watches := c.watches
 			c.watches = nil
@@ -80,7 +80,7 @@ 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":"initialize"'*) echo started >> "$STARTS"; 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
@@ -91,21 +91,28 @@ done
 // 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", ""},
+	oldMax, oldCall, oldFirst := maxLine, CallTimeout, RestartFirst
+	maxLine, RestartFirst = 1024, 10*time.Millisecond
+	t.Cleanup(func() { maxLine, CallTimeout, RestartFirst = oldMax, oldCall, oldFirst })
+	for _, row := range []struct {
+		mode    string
+		timeout time.Duration
+	}{
+		// The call's own bound is long: only the runner ending the bundle fails it within a second.
+		{"long", 10 * time.Second},
+		{"silent", 300 * time.Millisecond},
 	} {
 		t.Run(row.mode, func(t *testing.T) {
-			entry := filepath.Join(t.TempDir(), "bundle")
+			CallTimeout = row.timeout
+			dir := t.TempDir()
+			entry := filepath.Join(dir, "bundle")
+			starts := filepath.Join(dir, "starts")
 			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)
+			l, err := Start("caller", entry, append(os.Environ(), "MODE="+row.mode, "STARTS="+starts), &refusingState{}, logf)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -115,25 +122,41 @@ func TestAnOverlongLineAndAnUnansweredCallAreSaid(t *testing.T) {
 			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 {
+			want := "caller.t called by \"caller-1\""
+			if row.mode == "long" {
+				want = "ending the bundle"
+				if took := time.Since(begun); took > time.Second {
+					t.Fatalf("the call hung for %s", took)
+				}
+				if !strings.Contains(err.Error(), "over 1024 bytes") || strings.Contains(err.Error(), "did not answer") {
+					t.Fatalf("the call failed with %q, not the cap's reason", err)
+				}
+			}
+			for found := false; !found; {
 				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
-					}
+					found = strings.Contains(line, want)
 				case <-timeout:
 					t.Fatalf("not said; the log: %v", heard)
 				}
 			}
+			if row.mode != "long" {
+				return
+			}
+			// The bundle is started again: its start is written once more.
+			for {
+				if b, _ := os.ReadFile(starts); strings.Count(string(b), "started") >= 2 {
+					return
+				}
+				select {
+				case <-timeout:
+					t.Fatal("the bundle was not started again")
+				case <-time.After(10 * time.Millisecond):
+				}
+			}
 		})
 	}
 }