mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 10:45:59 -10:00
fix(plugin): keep a pinned tool's schema when a refresh drops it / 刷新丢弃工具时保留已固定的 schema
A server that drops a tool it had pinned into the provider-visible array was rewriting that array mid-session: the cache-stable prefix moved and every token behind it was spent again. The entry now stays until the session ends, the call it still invites fails with the live catalog's not-found, and the model is told on the turn tail — the one place an append leaves the prefix byte-stable. Registry.ReplacePrefix swaps a server's namespace under one lock. The removal and the re-registration were two calls, so a turn composing its request between them was built from a server that appeared to have no tools.
This commit is contained in:
1 parent
d0c057dbd9
commit
ca39c4b7b3
17 files changed
+864
-34
No files matched your search
@@ -19,6 +19,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/contract/config"
|
||||
"reasonix/internal/contract/event"
|
||||
"reasonix/internal/contract/provider"
|
||||
"reasonix/internal/session/control"
|
||||
@@ -126,10 +127,15 @@ model = "x"
|
||||
// finishes the turn on any round the script leaves out. Rounds are counted
|
||||
// across turns, so the script says which turn each call lands in.
|
||||
type listChangedProvider struct {
|
||||
mu sync.Mutex
|
||||
reqs []provider.Request
|
||||
mu sync.Mutex
|
||||
reqs []provider.Request
|
||||
// round counts provider requests across turns, so a script says which turn
|
||||
// each call lands in.
|
||||
round int
|
||||
calls map[int]string
|
||||
// calls are use_capability calls by round; direct are calls to a tool the
|
||||
// provider schema itself carries, which is the surface a pinned server has.
|
||||
calls map[int]string
|
||||
direct map[int]string
|
||||
}
|
||||
|
||||
func (p *listChangedProvider) Name() string { return "boot-mcp-list-changed" }
|
||||
@@ -139,15 +145,20 @@ func (p *listChangedProvider) Stream(_ context.Context, req provider.Request) (<
|
||||
p.reqs = append(p.reqs, req)
|
||||
round := p.round
|
||||
p.round++
|
||||
call := p.calls[round]
|
||||
call, direct := p.calls[round], p.direct[round]
|
||||
p.mu.Unlock()
|
||||
|
||||
ch := make(chan provider.Chunk, 2)
|
||||
if call != "" {
|
||||
switch {
|
||||
case call != "":
|
||||
ch <- provider.Chunk{Type: provider.ChunkToolCall, ToolCall: &provider.ToolCall{
|
||||
ID: fmt.Sprintf("probe-%d", round), Name: "use_capability", Arguments: call,
|
||||
}}
|
||||
} else {
|
||||
case direct != "":
|
||||
ch <- provider.Chunk{Type: provider.ChunkToolCall, ToolCall: &provider.ToolCall{
|
||||
ID: fmt.Sprintf("probe-%d", round), Name: direct, Arguments: "{}",
|
||||
}}
|
||||
default:
|
||||
ch <- provider.Chunk{Type: provider.ChunkText, Text: "ok"}
|
||||
}
|
||||
ch <- provider.Chunk{Type: provider.ChunkDone}
|
||||
@@ -262,3 +273,152 @@ func waitForRegisteredTool(t *testing.T, ctrl *control.Controller, name string)
|
||||
}
|
||||
t.Fatalf("%s never reached the registry after notifications/tools/list_changed", name)
|
||||
}
|
||||
|
||||
// A pinned server's tools are in the provider-visible array from the session's
|
||||
// first request, so one it drops would rewrite that array. It does not move:
|
||||
// the schema stays, its call fails with the typed not-found, the model is told
|
||||
// on the turn tail, and the next session never lists it. stdio only — each
|
||||
// Build spawns a helper from the same catalog; both transports are covered above.
|
||||
func TestEffectMCPListChangedKeepsAPinnedToolSurface(t *testing.T) {
|
||||
home := isolateConfigHome(t)
|
||||
reasonixHome := filepath.Join(home, ".reasonix")
|
||||
t.Setenv("REASONIX_HOME", reasonixHome)
|
||||
workspace := robustTempDir(t)
|
||||
t.Chdir(workspace)
|
||||
|
||||
kind := "boot-mcp-list-changed-pinned"
|
||||
var rec atomic.Pointer[listChangedProvider]
|
||||
provider.Register(kind, func(provider.Config) (provider.Provider, error) { return rec.Load(), nil })
|
||||
writeFile(t, workspace, "reasonix.toml", listChangedConfig(kind, listChangedServerBlock(t, "stdio")))
|
||||
approveWorkspace(t, workspace)
|
||||
approveProjectServer(t, workspace, "live")
|
||||
|
||||
// Session one only warms the schema cache: no tool is called, so the server
|
||||
// never announces and the catalog it cached is the one it started from.
|
||||
warm := &listChangedProvider{}
|
||||
rec.Store(warm)
|
||||
first, err := Build(t.Context(), Options{Sink: event.Discard, Home: reasonixHome, WorkspaceRoot: workspace})
|
||||
if err != nil {
|
||||
t.Fatalf("Build: %v", err)
|
||||
}
|
||||
waitForRegisteredTool(t, first, "mcp__live__beta")
|
||||
first.Close()
|
||||
waitForCachedMCPTool(t, "live", "beta")
|
||||
|
||||
// Session two is pinned: the server's tools are in the provider schema from
|
||||
// its first request, and it drops one mid-session.
|
||||
live := &listChangedProvider{direct: map[int]string{0: "mcp__live__alpha", 2: "mcp__live__beta"}}
|
||||
rec.Store(live)
|
||||
ctrl, err := Build(t.Context(), Options{Sink: event.Discard, Home: reasonixHome, WorkspaceRoot: workspace})
|
||||
if err != nil {
|
||||
t.Fatalf("Build: %v", err)
|
||||
}
|
||||
t.Cleanup(ctrl.Close)
|
||||
|
||||
if err := ctrl.Run(t.Context(), "call the pinned tool"); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
pinned := mcpToolNames(live.requests()[0])
|
||||
if !slices.Equal(pinned, []string{"mcp__live__alpha", "mcp__live__beta"}) {
|
||||
t.Fatalf("the first request carried %v, want the cached catalog pinned into the schema", pinned)
|
||||
}
|
||||
waitForRegisteredTool(t, ctrl, "mcp__live__gamma")
|
||||
|
||||
if err := ctrl.Run(t.Context(), "call the tool the server dropped"); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
requests := live.requests()
|
||||
last := requests[len(requests)-1]
|
||||
if moved := toolSurfaceDrift(requests[0], last); len(moved) != 0 {
|
||||
t.Fatalf("the provider-visible tools array moved with the catalog: %v", moved)
|
||||
}
|
||||
if got := toolResults(requests)["probe-2"]; !strings.Contains(got, `not found on server "live"`) {
|
||||
t.Fatalf("call to the withdrawn tool returned %q, want the typed not-found", got)
|
||||
}
|
||||
if !containsMessage(last, "withdrew mcp__live__beta") {
|
||||
t.Fatalf("the turn tail never told the model what the server withdrew:\n%s", strings.Join(messageTexts(last), "\n"))
|
||||
}
|
||||
if sys, after := systemMessage(requests[0].Messages), systemMessage(last.Messages); sys != after {
|
||||
t.Fatalf("the cache-stable prefix moved with the catalog:\n before: %.200q\n after: %.200q", sys, after)
|
||||
}
|
||||
|
||||
// Session three reads the refreshed cache: the withdrawn tool is gone from
|
||||
// the surface, and what the server added is pinned in its place.
|
||||
next := &listChangedProvider{}
|
||||
rec.Store(next)
|
||||
third, err := Build(t.Context(), Options{Sink: event.Discard, Home: reasonixHome, WorkspaceRoot: workspace})
|
||||
if err != nil {
|
||||
t.Fatalf("Build: %v", err)
|
||||
}
|
||||
t.Cleanup(third.Close)
|
||||
if err := third.Run(t.Context(), "say ok"); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
got := mcpToolNames(next.requests()[0])
|
||||
if !slices.Equal(got, []string{"mcp__live__alpha", "mcp__live__delta", "mcp__live__gamma"}) {
|
||||
t.Fatalf("the next session pinned %v, want the catalog the refresh cached", got)
|
||||
}
|
||||
}
|
||||
|
||||
// toolSurfaceDrift names the entries two requests do not share. The whole array
|
||||
// is compared — a changed description moves the prefix as a changed name does —
|
||||
// but only what moved is worth printing.
|
||||
func toolSurfaceDrift(before, after provider.Request) []string {
|
||||
entries := func(req provider.Request) map[string]string {
|
||||
out := make(map[string]string, len(req.Tools))
|
||||
for _, s := range req.Tools {
|
||||
out[s.Name] = s.Description + "|" + string(s.Parameters)
|
||||
}
|
||||
return out
|
||||
}
|
||||
was, is := entries(before), entries(after)
|
||||
var moved []string
|
||||
for name, body := range was {
|
||||
switch current, ok := is[name]; {
|
||||
case !ok:
|
||||
moved = append(moved, name+" (dropped)")
|
||||
case current != body:
|
||||
moved = append(moved, name+" (rewritten)")
|
||||
}
|
||||
}
|
||||
for name := range is {
|
||||
if _, ok := was[name]; !ok {
|
||||
moved = append(moved, name+" (added)")
|
||||
}
|
||||
}
|
||||
slices.Sort(moved)
|
||||
return moved
|
||||
}
|
||||
|
||||
func messageTexts(req provider.Request) []string {
|
||||
out := make([]string, 0, len(req.Messages))
|
||||
for _, m := range req.Messages {
|
||||
out = append(out, string(m.Role)+": "+m.Content)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func containsMessage(req provider.Request, want string) bool {
|
||||
for _, m := range req.Messages {
|
||||
if strings.Contains(m.Content, want) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// waitForCachedMCPTool waits for the handshake's schema write, which the next
|
||||
// session pins its provider surface from. The write is detached, so a session
|
||||
// holding the tools may not have cached them yet.
|
||||
func waitForCachedMCPTool(t *testing.T, server, tool string) {
|
||||
t.Helper()
|
||||
path := filepath.Join(config.CacheDir(), "mcp", server+".json")
|
||||
deadline := time.Now().Add(20 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if body, err := os.ReadFile(path); err == nil && strings.Contains(string(body), `"`+tool+`"`) {
|
||||
return
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("%s never reached the schema cache at %s", tool, path)
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
package tool
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
|
||||
"reasonix/internal/contract/provider"
|
||||
)
|
||||
|
||||
// ReplacePrefix swaps one name-prefixed namespace for the tools given, in a
|
||||
// single lock hold: a request composed between a RemovePrefix and its Add would
|
||||
// be built from a server that appears to have no tools. The admit ceiling and
|
||||
// suspended prefixes bind as they do in Add; a name outside prefix is ignored.
|
||||
// Reports how many tools the namespace holds afterwards.
|
||||
func (r *Registry) ReplacePrefix(prefix string, tools []Tool) int {
|
||||
if r == nil || prefix == "" {
|
||||
return 0
|
||||
}
|
||||
// admit may call back into a tool that takes its own lock, which a registry
|
||||
// swap must not be holding this one while it waits for.
|
||||
r.mu.RLock()
|
||||
admit := r.admit
|
||||
r.mu.RUnlock()
|
||||
|
||||
type entry struct {
|
||||
name string
|
||||
tool Tool
|
||||
canon json.RawMessage
|
||||
}
|
||||
admitted := make([]entry, 0, len(tools))
|
||||
for _, t := range tools {
|
||||
if t == nil {
|
||||
continue
|
||||
}
|
||||
name := t.Name()
|
||||
if !strings.HasPrefix(name, prefix) {
|
||||
continue
|
||||
}
|
||||
if admit != nil && !admit(t) {
|
||||
continue
|
||||
}
|
||||
admitted = append(admitted, entry{name: name, tool: t, canon: provider.CanonicalizeSchema(t.Schema())})
|
||||
}
|
||||
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
kept := r.order[:0]
|
||||
removed := 0
|
||||
for _, name := range r.order {
|
||||
if strings.HasPrefix(name, prefix) {
|
||||
delete(r.tools, name)
|
||||
delete(r.canon, name)
|
||||
removed++
|
||||
continue
|
||||
}
|
||||
kept = append(kept, name)
|
||||
}
|
||||
r.order = kept
|
||||
registered := 0
|
||||
for _, e := range admitted {
|
||||
if r.suspendedLocked(e.name) {
|
||||
continue
|
||||
}
|
||||
if _, ok := r.tools[e.name]; !ok {
|
||||
r.order = append(r.order, e.name)
|
||||
}
|
||||
r.tools[e.name] = e.tool
|
||||
r.canon[e.name] = e.canon
|
||||
registered++
|
||||
}
|
||||
if removed > 0 || registered > 0 {
|
||||
r.schemaRev.Add(1)
|
||||
}
|
||||
return registered
|
||||
}
|
||||
|
||||
// suspendedLocked reports whether name falls under a prefix this session turned
|
||||
// off. Callers hold r.mu.
|
||||
func (r *Registry) suspendedLocked(name string) bool {
|
||||
for prefix := range r.suspended {
|
||||
if strings.HasPrefix(name, prefix) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
package tool
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The window a remove-then-add pair leaves open is what this closes: a request
|
||||
// composed inside it would be built from a server with no tools at all, which
|
||||
// is both a moved cache prefix and a call the model was offered resolving to
|
||||
// nothing.
|
||||
func TestReplacePrefixNeverShowsAnEmptyNamespace(t *testing.T) {
|
||||
r := NewRegistry()
|
||||
first := []Tool{stubTool{name: "mcp__atlas__a"}, stubTool{name: "mcp__atlas__b"}}
|
||||
second := []Tool{stubTool{name: "mcp__atlas__b"}, stubTool{name: "mcp__atlas__c"}}
|
||||
r.ReplacePrefix("mcp__atlas__", first)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
stop := make(chan struct{})
|
||||
lowest := 2
|
||||
var mu sync.Mutex
|
||||
for range 4 {
|
||||
wg.Go(func() {
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
default:
|
||||
}
|
||||
seen := 0
|
||||
for _, s := range r.Schemas() {
|
||||
if strings.HasPrefix(s.Name, "mcp__atlas__") {
|
||||
seen++
|
||||
}
|
||||
}
|
||||
mu.Lock()
|
||||
lowest = min(lowest, seen)
|
||||
mu.Unlock()
|
||||
}
|
||||
})
|
||||
}
|
||||
for i := range 200 {
|
||||
if i%2 == 0 {
|
||||
r.ReplacePrefix("mcp__atlas__", second)
|
||||
continue
|
||||
}
|
||||
r.ReplacePrefix("mcp__atlas__", first)
|
||||
}
|
||||
close(stop)
|
||||
wg.Wait()
|
||||
|
||||
if lowest < 2 {
|
||||
t.Fatalf("a reader saw %d tools under the prefix mid-swap, want never fewer than 2", lowest)
|
||||
}
|
||||
}
|
||||
|
||||
// Suspension is the user turning a server off. A swap is not a way back in.
|
||||
func TestReplacePrefixLeavesASuspendedPrefixOff(t *testing.T) {
|
||||
r := NewRegistry()
|
||||
r.SuspendPrefix("mcp__atlas__")
|
||||
if got := r.ReplacePrefix("mcp__atlas__", []Tool{stubTool{name: "mcp__atlas__a"}}); got != 0 {
|
||||
t.Fatalf("registered %d tools into a suspended prefix, want 0", got)
|
||||
}
|
||||
if _, ok := r.Get("mcp__atlas__a"); ok {
|
||||
t.Fatal("a swap registered a tool the session had turned off")
|
||||
}
|
||||
}
|
||||
|
||||
// The admit ceiling is what a restricted registry may ever hold. No
|
||||
// registration path may widen it, this one included.
|
||||
func TestReplacePrefixHonorsTheAdmitCeiling(t *testing.T) {
|
||||
r := NewRegistry()
|
||||
r.Restrict(func(tl Tool) bool { return tl.Name() != "mcp__atlas__b" })
|
||||
r.ReplacePrefix("mcp__atlas__", []Tool{stubTool{name: "mcp__atlas__a"}, stubTool{name: "mcp__atlas__b"}})
|
||||
|
||||
if _, ok := r.Get("mcp__atlas__a"); !ok {
|
||||
t.Fatal("an admitted tool did not reach the registry")
|
||||
}
|
||||
if _, ok := r.Get("mcp__atlas__b"); ok {
|
||||
t.Fatal("a swap admitted a tool the ceiling rejects")
|
||||
}
|
||||
}
|
||||
|
||||
// The namespace is what the call owns: a tool from somewhere else would be
|
||||
// registered here and then removed by the next swap of a prefix it never had.
|
||||
func TestReplacePrefixIgnoresToolsOutsideThePrefix(t *testing.T) {
|
||||
r := NewRegistry()
|
||||
r.Add(stubTool{name: "mcp__ledger__post"})
|
||||
r.ReplacePrefix("mcp__atlas__", []Tool{stubTool{name: "mcp__ledger__other"}})
|
||||
|
||||
if _, ok := r.Get("mcp__ledger__other"); ok {
|
||||
t.Fatal("a swap registered a tool outside the prefix it owns")
|
||||
}
|
||||
if _, ok := r.Get("mcp__ledger__post"); !ok {
|
||||
t.Fatal("a swap removed a tool outside the prefix it owns")
|
||||
}
|
||||
}
|
||||
@@ -440,6 +440,20 @@ func (r *Registry) ProviderVisible(name string) bool {
|
||||
return r.providerVisible[strings.TrimSpace(name)]
|
||||
}
|
||||
|
||||
// ProviderSurfacePinned reports whether name belongs to an explicit
|
||||
// provider-visible allowlist. Unlike ProviderVisible it is false when no
|
||||
// allowlist is set: a registry that shows whatever it holds promises nothing
|
||||
// about which names a provider sees, so nothing has to be held back to keep
|
||||
// that promise.
|
||||
func (r *Registry) ProviderSurfacePinned(name string) bool {
|
||||
if r == nil {
|
||||
return false
|
||||
}
|
||||
r.mu.RLock()
|
||||
defer r.mu.RUnlock()
|
||||
return r.providerVisible != nil && r.providerVisible[strings.TrimSpace(name)]
|
||||
}
|
||||
|
||||
func (r *Registry) isProviderVisibleLocked(name string) bool {
|
||||
if r.providerVisible == nil {
|
||||
return true
|
||||
@@ -461,10 +475,8 @@ func (r *Registry) Add(t Tool) {
|
||||
defer r.mu.Unlock()
|
||||
|
||||
name := t.Name()
|
||||
for prefix := range r.suspended {
|
||||
if strings.HasPrefix(name, prefix) {
|
||||
return
|
||||
}
|
||||
if r.suspendedLocked(name) {
|
||||
return
|
||||
}
|
||||
if _, ok := r.tools[name]; !ok {
|
||||
r.order = append(r.order, name)
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
package tool
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
)
|
||||
|
||||
// withdrawnTool holds a provider schema slot open for a tool that is gone: the
|
||||
// visible array is fixed for the session, so vacating an entry rewrites the
|
||||
// cached prefix and costs every token behind it. The entry stays byte for byte
|
||||
// and its call fails with the reason. Embedding the Tool interface strips the
|
||||
// rest — an MCP binding, a preview — so catalogs stop offering it as live.
|
||||
type withdrawnTool struct {
|
||||
Tool
|
||||
reason string
|
||||
}
|
||||
|
||||
// Withdraw returns t as a schema-only placeholder that fails with reason.
|
||||
// Withdrawing an already-withdrawn tool restates the reason rather than
|
||||
// wrapping it twice.
|
||||
func Withdraw(t Tool, reason string) Tool {
|
||||
if t == nil {
|
||||
return nil
|
||||
}
|
||||
if w, ok := t.(withdrawnTool); ok {
|
||||
return withdrawnTool{Tool: w.Tool, reason: reason}
|
||||
}
|
||||
return withdrawnTool{Tool: t, reason: reason}
|
||||
}
|
||||
|
||||
// IsWithdrawn reports whether t is only holding a schema slot. Callers that
|
||||
// route by what a tool can still do read this rather than calling it to find out.
|
||||
func IsWithdrawn(t Tool) bool {
|
||||
_, ok := t.(withdrawnTool)
|
||||
return ok
|
||||
}
|
||||
|
||||
func (w withdrawnTool) Execute(context.Context, json.RawMessage) (string, error) {
|
||||
return "", errors.New(w.reason)
|
||||
}
|
||||
|
||||
// ReadOnly is true because the call only states its reason: approving a
|
||||
// withdrawn tool would be approving an error message.
|
||||
func (w withdrawnTool) ReadOnly() bool { return true }
|
||||
@@ -0,0 +1,63 @@
|
||||
package tool
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// Withdrawing has to leave the provider-visible array byte-identical: the point
|
||||
// of keeping the slot is that the cached prefix does not move.
|
||||
func TestWithdrawKeepsTheSchemaAndFailsTheCall(t *testing.T) {
|
||||
r := NewRegistry()
|
||||
live := stubTool{name: "mcp__atlas__gone", server: "atlas", raw: "gone"}
|
||||
r.Add(live)
|
||||
before := r.Schemas()
|
||||
|
||||
r.Add(Withdraw(live, `MCP tool "gone" not found on server "atlas"`))
|
||||
after := r.Schemas()
|
||||
|
||||
if len(before) != 1 || len(after) != 1 || !reflect.DeepEqual(before[0], after[0]) {
|
||||
t.Fatalf("the schema moved when the tool was withdrawn:\n before: %+v\n after: %+v", before, after)
|
||||
}
|
||||
held, _ := r.Get("mcp__atlas__gone")
|
||||
if !IsWithdrawn(held) {
|
||||
t.Fatal("the registry holds a live tool where a withdrawn slot was registered")
|
||||
}
|
||||
if _, err := held.Execute(t.Context(), nil); err == nil || !strings.Contains(err.Error(), "not found on server") {
|
||||
t.Fatalf("calling a withdrawn tool returned %v, want its reason", err)
|
||||
}
|
||||
if !held.ReadOnly() {
|
||||
t.Fatal("a withdrawn tool reported a side effect it cannot have")
|
||||
}
|
||||
}
|
||||
|
||||
// A slot is not a capability: anything that routes by what a tool can still do
|
||||
// has to stop seeing this one, or a catalog keeps offering a dead call.
|
||||
func TestWithdrawDropsTheMCPBinding(t *testing.T) {
|
||||
r := NewRegistry()
|
||||
live := stubTool{name: "mcp__atlas__gone", server: "atlas", raw: "gone"}
|
||||
r.Add(Withdraw(live, "gone"))
|
||||
|
||||
if got := r.MCPBindings(); len(got) != 0 {
|
||||
t.Fatalf("a withdrawn tool still publishes %d MCP bindings, want none", len(got))
|
||||
}
|
||||
if IsWithdrawn(live) {
|
||||
t.Fatal("a live tool reported itself withdrawn")
|
||||
}
|
||||
}
|
||||
|
||||
// Restating a reason must not wrap a slot inside another slot: the second
|
||||
// reason is the current one, and the schema it holds open is still the
|
||||
// original tool's.
|
||||
func TestWithdrawRestatesRatherThanNests(t *testing.T) {
|
||||
once := Withdraw(stubTool{name: "mcp__atlas__gone"}, "first")
|
||||
twice := Withdraw(once, "second")
|
||||
|
||||
if _, err := twice.Execute(t.Context(), nil); err == nil || err.Error() != "second" {
|
||||
t.Fatalf("reason = %v, want the restated one", err)
|
||||
}
|
||||
if twice.Name() != "mcp__atlas__gone" {
|
||||
t.Fatalf("name = %q, want the original tool's", twice.Name())
|
||||
}
|
||||
}
|
||||
@@ -28,8 +28,11 @@ type listChangedServer struct {
|
||||
// selfNotify stands for a server that announces on every listing, which is
|
||||
// what bounded catch-up exists for.
|
||||
selfNotify bool
|
||||
lists atomic.Int64
|
||||
calls atomic.Int64
|
||||
// destructive names the tools the server annotates as destructive, which is
|
||||
// one of the facts the host's gate reads before a call runs.
|
||||
destructive map[string]bool
|
||||
lists atomic.Int64
|
||||
calls atomic.Int64
|
||||
}
|
||||
|
||||
func newListChangedServer(t *testing.T, declare bool, tools ...string) (*listChangedServer, *httptest.Server) {
|
||||
@@ -102,11 +105,15 @@ func (s *listChangedServer) catalog() []map[string]any {
|
||||
defer s.mu.Unlock()
|
||||
out := make([]map[string]any, 0, len(s.tools))
|
||||
for _, name := range s.tools {
|
||||
out = append(out, map[string]any{
|
||||
entry := map[string]any{
|
||||
"name": name,
|
||||
"description": "tool " + name,
|
||||
"inputSchema": map[string]any{"type": "object"},
|
||||
})
|
||||
}
|
||||
if s.destructive[name] {
|
||||
entry["annotations"] = map[string]any{"destructiveHint": true}
|
||||
}
|
||||
out = append(out, entry)
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -473,3 +480,75 @@ func TestToolListChangedRewritesTheSchemaCache(t *testing.T) {
|
||||
t.Fatalf("cached catalog = %v, want the one the server moved to", names)
|
||||
}
|
||||
}
|
||||
|
||||
// A connected server can add a tool after startup, which is the boundary this
|
||||
// change moves. What arrives has to reach the host's gate carrying the same
|
||||
// facts a startup tool does — the server's authorization and its own
|
||||
// annotations — or a tool could get past a check by arriving late.
|
||||
func TestToolListChangedCarriesTheStartupSecurityFacts(t *testing.T) {
|
||||
server, srv := newListChangedServer(t, true, "alpha")
|
||||
server.destructive = map[string]bool{"gamma": true}
|
||||
h, published := connectListChanged(t, srv, Spec{Authorized: true})
|
||||
|
||||
startup, err := h.ToolsFor(t.Context(), "live")
|
||||
if err != nil {
|
||||
t.Fatalf("ToolsFor: %v", err)
|
||||
}
|
||||
if len(startup) != 1 || !tool.IsMCPServerAuthorized(startup[0]) {
|
||||
t.Fatalf("startup tool authorization = %v, want the spec's", names(startup))
|
||||
}
|
||||
|
||||
server.announce(1, "alpha", "gamma")
|
||||
if _, err := h.lookupClient("live").call(t.Context(), "tools/call",
|
||||
map[string]any{"name": "alpha", "arguments": map[string]any{}}); err != nil {
|
||||
t.Fatalf("tools/call: %v", err)
|
||||
}
|
||||
waitForCatalog(t, published)
|
||||
|
||||
refreshed, err := h.ToolsFor(t.Context(), "live")
|
||||
if err != nil {
|
||||
t.Fatalf("ToolsFor after refresh: %v", err)
|
||||
}
|
||||
byName := map[string]tool.Tool{}
|
||||
for _, tl := range refreshed {
|
||||
byName[tl.Name()] = tl
|
||||
}
|
||||
arrived, ok := byName["mcp__live__gamma"]
|
||||
if !ok {
|
||||
t.Fatalf("the added tool never reached the catalog: %v", names(refreshed))
|
||||
}
|
||||
if !tool.IsMCPServerAuthorized(arrived) {
|
||||
t.Fatal("a tool that arrived mid-session lost the server authorization its siblings carry")
|
||||
}
|
||||
if !tool.HasMCPDestructiveHint(arrived) {
|
||||
t.Fatal("a tool that arrived mid-session lost the destructive hint it declared")
|
||||
}
|
||||
if tool.HasMCPDestructiveHint(byName["mcp__live__alpha"]) {
|
||||
t.Fatal("the refresh marked an undeclared tool destructive")
|
||||
}
|
||||
}
|
||||
|
||||
// An unauthorized server stays unauthorized for whatever it adds later: the
|
||||
// approval is the user's answer about the server, and a server cannot answer
|
||||
// it for them by announcing a new tool.
|
||||
func TestToolListChangedCannotSelfAuthorize(t *testing.T) {
|
||||
server, srv := newListChangedServer(t, true, "alpha")
|
||||
h, published := connectListChanged(t, srv, Spec{})
|
||||
|
||||
server.announce(1, "alpha", "gamma")
|
||||
if _, err := h.lookupClient("live").call(t.Context(), "tools/call",
|
||||
map[string]any{"name": "alpha", "arguments": map[string]any{}}); err != nil {
|
||||
t.Fatalf("tools/call: %v", err)
|
||||
}
|
||||
waitForCatalog(t, published)
|
||||
|
||||
refreshed, err := h.ToolsFor(t.Context(), "live")
|
||||
if err != nil {
|
||||
t.Fatalf("ToolsFor after refresh: %v", err)
|
||||
}
|
||||
for _, tl := range refreshed {
|
||||
if tool.IsMCPServerAuthorized(tl) {
|
||||
t.Fatalf("%s reported an authorization the user never gave", tl.Name())
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -285,6 +285,10 @@ type Agent struct {
|
||||
// is unavoidable but limited to one call — the prefix stays stable otherwise.
|
||||
steer steerInbox
|
||||
|
||||
// hostNotices is runtime-authored fact the next round appends to the turn
|
||||
// tail. See host_notice.go.
|
||||
hostNotices hostNoticeQueue
|
||||
|
||||
// task is the state shared by every Run continuing one delivery scope: the
|
||||
// receipt ledger complete_step validates citations against, the spend that
|
||||
// outlives a single Run, and the guards keyed to the task rather than the
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
// host_notice.go — what the host has to tell the model about state that moved
|
||||
// under it, delivered on the turn tail.
|
||||
package agent
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"reasonix/internal/contract/event"
|
||||
"reasonix/internal/contract/provider"
|
||||
"reasonix/internal/state/sessionstore"
|
||||
)
|
||||
|
||||
// maxPendingHostNotices bounds what a host that keeps changing its mind can
|
||||
// append to one turn. A notice the model never reads is a notice it does not
|
||||
// need: the surface it describes is readable at any time.
|
||||
const maxPendingHostNotices = 8
|
||||
|
||||
// hostNoticeQueue is runtime-authored fact waiting for a round boundary. Unlike
|
||||
// steerInbox it accepts while no turn is running: the fact is true from the
|
||||
// moment it is queued, and the next turn is the first one that can act on it.
|
||||
type hostNoticeQueue struct {
|
||||
mu sync.Mutex
|
||||
pending []string
|
||||
}
|
||||
|
||||
// NoteHostFact queues one sentence for the next round's turn tail. It is
|
||||
// appended after the cached prefix, never into it, so telling the model costs
|
||||
// no prefix rewrite. A fact already waiting is not repeated.
|
||||
func (a *Agent) NoteHostFact(text string) {
|
||||
text = strings.TrimSpace(text)
|
||||
if a == nil || text == "" {
|
||||
return
|
||||
}
|
||||
a.hostNotices.mu.Lock()
|
||||
defer a.hostNotices.mu.Unlock()
|
||||
if slices.Contains(a.hostNotices.pending, text) {
|
||||
return
|
||||
}
|
||||
if len(a.hostNotices.pending) >= maxPendingHostNotices {
|
||||
return
|
||||
}
|
||||
a.hostNotices.pending = append(a.hostNotices.pending, text)
|
||||
}
|
||||
|
||||
// takeHostFacts hands over everything queued, leaving the queue empty.
|
||||
func (a *Agent) takeHostFacts() []string {
|
||||
if a == nil {
|
||||
return nil
|
||||
}
|
||||
a.hostNotices.mu.Lock()
|
||||
defer a.hostNotices.mu.Unlock()
|
||||
pending := a.hostNotices.pending
|
||||
a.hostNotices.pending = nil
|
||||
return pending
|
||||
}
|
||||
|
||||
// appendTurnTailNotices states what the host knows and the model does not, on
|
||||
// the turn tail rather than in the cached prefix: an append leaves the prefix
|
||||
// byte-stable, so telling the model what moved under the session costs no
|
||||
// cache miss.
|
||||
func (a *Agent) appendTurnTailNotices() {
|
||||
for _, fact := range a.takeHostFacts() {
|
||||
a.sess.conversation.Add(provider.Message{Role: provider.RoleUser, Content: sessionstore.MidTurnSteerMessage(fact, true)})
|
||||
a.svc.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelInfo, Text: fact})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"slices"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// A fact queued between turns still has to be delivered: the state it describes
|
||||
// moved while nothing was running, and the next turn is the first one that can
|
||||
// act on it. That is what separates this from a steer.
|
||||
func TestHostFactsWaitForTheNextRound(t *testing.T) {
|
||||
a := &Agent{}
|
||||
a.NoteHostFact("the live server withdrew mcp__live__beta")
|
||||
a.NoteHostFact(" the live server withdrew mcp__live__beta ")
|
||||
a.NoteHostFact("")
|
||||
|
||||
got := a.takeHostFacts()
|
||||
if !slices.Equal(got, []string{"the live server withdrew mcp__live__beta"}) {
|
||||
t.Fatalf("pending facts = %v, want the one fact stated once", got)
|
||||
}
|
||||
if got := a.takeHostFacts(); len(got) != 0 {
|
||||
t.Fatalf("facts were delivered twice: %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// A host that keeps changing its mind must not be able to grow one turn's tail
|
||||
// without bound.
|
||||
func TestHostFactsAreBounded(t *testing.T) {
|
||||
a := &Agent{}
|
||||
for i := range maxPendingHostNotices + 5 {
|
||||
a.NoteHostFact(fmt.Sprintf("fact %d", i))
|
||||
}
|
||||
if got := len(a.takeHostFacts()); got != maxPendingHostNotices {
|
||||
t.Fatalf("queued %d facts, want the ceiling of %d", got, maxPendingHostNotices)
|
||||
}
|
||||
}
|
||||
@@ -276,6 +276,8 @@ func (a *Agent) runToolLoop(ctx context.Context, state *turnRuntime) error {
|
||||
// (unapplied path marks uncertain + pause via the notice sink).
|
||||
a.recordUnappliedSteer("(body load failed)", entry.host, entry.itemID)
|
||||
}
|
||||
a.appendTurnTailNotices()
|
||||
|
||||
schemas := a.svc.tools.ProviderSchemas(ctx)
|
||||
a.sess.lastProviderSchemas = schemas
|
||||
prefixShape := a.capturePrefixShape(schemas)
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
package usecap
|
||||
|
||||
import "reasonix/internal/contract/tool"
|
||||
|
||||
// routableRegistryTool is the registry hit a capability call may bind to. A
|
||||
// withdrawn entry is a slot held open for a tool the server dropped, not a
|
||||
// target: the live catalog decides, so a tool that comes back is callable.
|
||||
func routableRegistryTool(reg *tool.Registry, name string) (tool.Tool, bool) {
|
||||
tl, ok := reg.Get(name)
|
||||
return tl, ok && !tool.IsWithdrawn(tl)
|
||||
}
|
||||
@@ -909,7 +909,7 @@ func (t *UseCapabilityTool) resolveCall(ctx context.Context, id string, args jso
|
||||
// let them silently miss.
|
||||
modelName := plugin.ModelToolName(server, raw)
|
||||
if t.registry != nil {
|
||||
if tl, ok := t.registry.Get(modelName); ok {
|
||||
if tl, ok := routableRegistryTool(t.registry, modelName); ok {
|
||||
if t.runtime != nil && !plugin.MCPToolMatchesSpec(tl, runtimeSpec) {
|
||||
return t.resolveUnavailable(base, id, modelName, fmt.Sprintf("connected MCP server %q identity does not match the current runtime configuration", server)), nil
|
||||
}
|
||||
|
||||
@@ -179,7 +179,7 @@ func newControllerDeps(opts Options, sink event.Sink, usageTee *goalUsageTee, ru
|
||||
jobs: opts.Jobs,
|
||||
taskStore: opts.TaskStore,
|
||||
workspaceLease: opts.WorkspaceLease,
|
||||
mcp: newMcpManager(opts.Host, opts.Registry, pluginCtx, opts.MCPDefaultCallTimeout),
|
||||
mcp: newMcpManager(opts.Host, opts.Registry, pluginCtx, opts.MCPDefaultCallTimeout, hostFactNotice(opts.Executor)),
|
||||
mcpConfigureSpec: opts.MCPConfigureSpec,
|
||||
capabilityRuntime: opts.CapabilityRuntime,
|
||||
ablation: opts.Ablation,
|
||||
|
||||
@@ -4,11 +4,13 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/contract/tool"
|
||||
"reasonix/internal/ext/plugin"
|
||||
"reasonix/internal/runtime/agent"
|
||||
)
|
||||
|
||||
// mcpManager owns the session's live tool/plugin surface: the MCP plugin Host
|
||||
@@ -35,37 +37,93 @@ type mcpManager struct {
|
||||
// controller is handed to a caller and only read after.
|
||||
sealed error
|
||||
promptFailures promptFailureDebt
|
||||
// notice tells the model what a catalog refresh took away. It is the
|
||||
// executor's turn tail, bound once at construction.
|
||||
notice func(string)
|
||||
}
|
||||
|
||||
// seal makes the manager refuse to connect or register any server.
|
||||
func (m *mcpManager) seal(reason error) { m.sealed = reason }
|
||||
|
||||
func newMcpManager(host *plugin.Host, reg *tool.Registry, pluginCtx context.Context, defaultCallTimeout time.Duration) mcpManager {
|
||||
watchToolListChanges(host, reg, pluginCtx)
|
||||
return mcpManager{host: host, reg: reg, pluginCtx: pluginCtx, defaultCallTimeout: defaultCallTimeout}
|
||||
func newMcpManager(host *plugin.Host, reg *tool.Registry, pluginCtx context.Context, defaultCallTimeout time.Duration, notice func(string)) mcpManager {
|
||||
watchToolListChanges(host, reg, pluginCtx, notice)
|
||||
return mcpManager{host: host, reg: reg, pluginCtx: pluginCtx, defaultCallTimeout: defaultCallTimeout, notice: notice}
|
||||
}
|
||||
|
||||
// watchToolListChanges keeps this session's registry in step with a server that
|
||||
// announces a changed tool list. The host may be shared between sessions, so
|
||||
// each one registers the refreshed catalog into its own registry.
|
||||
func watchToolListChanges(host *plugin.Host, reg *tool.Registry, pluginCtx context.Context) {
|
||||
// each one registers the refreshed catalog into its own registry. notice, when
|
||||
// set, is how the session tells the model what the refresh took away from it.
|
||||
func watchToolListChanges(host *plugin.Host, reg *tool.Registry, pluginCtx context.Context, notice func(string)) {
|
||||
if host == nil || reg == nil {
|
||||
return
|
||||
}
|
||||
host.SubscribeToolListChanges(pluginCtx, func(spec plugin.Spec, tools []tool.Tool) {
|
||||
registerRefreshedMCPTools(reg, spec, tools)
|
||||
withdrawn := registerRefreshedMCPTools(reg, spec, tools)
|
||||
if len(withdrawn) > 0 && notice != nil {
|
||||
notice(withdrawnMCPToolsNotice(spec.Name, withdrawn))
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// registerRefreshedMCPTools swaps one server's tools for the set it now offers.
|
||||
// The prefix is deliberately not resumed: a session that suspended this server
|
||||
// turned it off, and a server announcing new tools is not the user asking for
|
||||
// it back.
|
||||
func registerRefreshedMCPTools(reg *tool.Registry, spec plugin.Spec, tools []tool.Tool) {
|
||||
reg.RemovePrefix(plugin.ToolPrefix(spec.Name))
|
||||
for _, t := range tools {
|
||||
reg.Add(t)
|
||||
// registerRefreshedMCPTools swaps one server's tools for the set it now offers,
|
||||
// in one registry call, and reports the pinned names the catalog no longer
|
||||
// backs — those stay as withdrawn slots (see withdrawnProviderTools). The
|
||||
// prefix is not resumed: a session that suspended this server turned it off,
|
||||
// and a server announcing new tools is not the user asking for it back.
|
||||
func registerRefreshedMCPTools(reg *tool.Registry, spec plugin.Spec, tools []tool.Tool) []string {
|
||||
prefix := plugin.ToolPrefix(spec.Name)
|
||||
held, withdrawn := withdrawnProviderTools(reg, spec.Name, prefix, tools)
|
||||
reg.ReplacePrefix(prefix, append(slices.Clone(tools), held...))
|
||||
return withdrawn
|
||||
}
|
||||
|
||||
// withdrawnProviderTools are the slots the refresh must not vacate: what this
|
||||
// session pinned into the provider-visible array at boot and the catalog no
|
||||
// longer offers. Vacating one rewrites the cache-stable prefix, so the entry
|
||||
// stays and its call fails with the live catalog's not-found. A slot an earlier
|
||||
// refresh withdrew is kept but not announced twice.
|
||||
func withdrawnProviderTools(reg *tool.Registry, server, prefix string, offered []tool.Tool) ([]tool.Tool, []string) {
|
||||
live := make(map[string]bool, len(offered))
|
||||
for _, t := range offered {
|
||||
if t != nil {
|
||||
live[t.Name()] = true
|
||||
}
|
||||
}
|
||||
raw := map[string]string{}
|
||||
for _, b := range reg.MCPBindings() {
|
||||
raw[b.CallableName] = b.RawName
|
||||
}
|
||||
var held []tool.Tool
|
||||
var withdrawn []string
|
||||
for _, name := range reg.AllNames() {
|
||||
if !strings.HasPrefix(name, prefix) || live[name] || !reg.ProviderSurfacePinned(name) {
|
||||
continue
|
||||
}
|
||||
current, ok := reg.Get(name)
|
||||
if !ok || current == nil {
|
||||
continue
|
||||
}
|
||||
if tool.IsWithdrawn(current) {
|
||||
held = append(held, current)
|
||||
continue
|
||||
}
|
||||
rawName := raw[name]
|
||||
if rawName == "" {
|
||||
rawName = strings.TrimPrefix(name, prefix)
|
||||
}
|
||||
held = append(held, tool.Withdraw(current, fmt.Sprintf("MCP tool %q not found on server %q", rawName, server)))
|
||||
withdrawn = append(withdrawn, name)
|
||||
}
|
||||
return held, withdrawn
|
||||
}
|
||||
|
||||
// withdrawnMCPToolsNotice is what the model is told on the turn tail: the tools
|
||||
// array still shows these calls and will until the session ends, so a model
|
||||
// reading the array and nothing else would keep spending turns on them.
|
||||
func withdrawnMCPToolsNotice(server string, names []string) string {
|
||||
return fmt.Sprintf("MCP server %q withdrew %s during this session. The definition stays in your tool list so the cached prefix does not move, but calling it now fails with a not-found; use what the server offers instead (use_capability inspect mcp-server:%s lists it).",
|
||||
server, strings.Join(names, ", "), server)
|
||||
}
|
||||
|
||||
// hostRef returns the live plugin host (nil until one is injected or lazily
|
||||
@@ -156,7 +214,7 @@ func (m *mcpManager) ensureHost() (*plugin.Host, context.Context, *tool.Registry
|
||||
host, ctx, reg := m.host, m.pluginCtx, m.reg
|
||||
m.mu.Unlock()
|
||||
if created {
|
||||
watchToolListChanges(host, reg, ctx)
|
||||
watchToolListChanges(host, reg, ctx, m.notice)
|
||||
}
|
||||
return host, ctx, reg
|
||||
}
|
||||
@@ -273,3 +331,13 @@ func (m *mcpManager) readResource(ctx context.Context, server, uri string) (stri
|
||||
}
|
||||
return h.ReadResource(ctx, server, uri)
|
||||
}
|
||||
|
||||
// hostFactNotice is how a background change reaches the model: the executor's
|
||||
// turn tail, which an append leaves the cached prefix alone. A session with no
|
||||
// executor has nowhere to say it, and says nothing.
|
||||
func hostFactNotice(executor *agent.Agent) func(string) {
|
||||
if executor == nil {
|
||||
return nil
|
||||
}
|
||||
return executor.NoteHostFact
|
||||
}
|
||||
@@ -37,7 +37,7 @@ func TestCatalogToolsCountsWhatIsCallableRatherThanWhatIsRunning(t *testing.T) {
|
||||
reg.Add(catalogTool{name: "mcp__atlas__rewrite", server: "atlas", raw: "rewrite"})
|
||||
reg.Add(catalogTool{name: "mcp__ledger__post", server: "ledger", raw: "post"})
|
||||
|
||||
c := &Controller{controllerDeps: controllerDeps{mcp: newMcpManager(nil, reg, t.Context(), 0)}}
|
||||
c := &Controller{controllerDeps: controllerDeps{mcp: newMcpManager(nil, reg, t.Context(), 0, nil)}}
|
||||
got := c.MCPCatalogTools()
|
||||
|
||||
if got["atlas"] != 2 {
|
||||
@@ -54,7 +54,7 @@ func TestCatalogToolsCountsWhatIsCallableRatherThanWhatIsRunning(t *testing.T) {
|
||||
// No registry is not an empty registry: a caller that cannot tell them apart
|
||||
// would report every configured server as having nothing to offer.
|
||||
func TestCatalogToolsAnswersNothingWithoutARegistry(t *testing.T) {
|
||||
c := &Controller{controllerDeps: controllerDeps{mcp: newMcpManager(nil, nil, t.Context(), 0)}}
|
||||
c := &Controller{controllerDeps: controllerDeps{mcp: newMcpManager(nil, nil, t.Context(), 0, nil)}}
|
||||
if got := c.MCPCatalogTools(); got != nil {
|
||||
t.Fatalf("catalog = %v with no registry, want nil", got)
|
||||
}
|
||||
@@ -69,7 +69,7 @@ func TestDisconnectClearsDisabledMCPPolicy(t *testing.T) {
|
||||
CallableName: "mcp__mock__write",
|
||||
CapabilityID: "mcp-tool:mock/write",
|
||||
}})
|
||||
m := newMcpManager(nil, reg, context.Background(), 0)
|
||||
m := newMcpManager(nil, reg, context.Background(), 0, nil)
|
||||
|
||||
m.disconnect("mock")
|
||||
if reg.DisabledMCP("mcp__mock__write") || reg.DisabledMCP("mcp-tool:mock/write") {
|
||||
|
||||
@@ -1,8 +1,11 @@
|
||||
package control
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"reasonix/internal/contract/provider"
|
||||
"reasonix/internal/contract/tool"
|
||||
"reasonix/internal/ext/plugin"
|
||||
)
|
||||
@@ -51,3 +54,97 @@ func TestRefreshedMCPToolsLeaveASuspendedServerOff(t *testing.T) {
|
||||
t.Fatal("a refresh brought back a server this session had turned off")
|
||||
}
|
||||
}
|
||||
|
||||
// The provider-visible tools array is fixed for the session: a pinned server
|
||||
// that drops a tool must not take its schema out of the array, or every cached
|
||||
// token behind it is spent again on the next request.
|
||||
func TestRefreshedMCPToolsKeepAPinnedSchemaSlot(t *testing.T) {
|
||||
reg := tool.NewRegistry()
|
||||
reg.Add(refreshedTool("mcp__atlas__gone", "gone"))
|
||||
reg.Add(refreshedTool("mcp__atlas__kept", "kept"))
|
||||
reg.SetProviderVisibleTools([]string{"mcp__atlas__gone", "mcp__atlas__kept"})
|
||||
before := reg.Schemas()
|
||||
|
||||
withdrawn := registerRefreshedMCPTools(reg, plugin.Spec{Name: "atlas"}, []tool.Tool{
|
||||
refreshedTool("mcp__atlas__kept", "kept"),
|
||||
refreshedTool("mcp__atlas__added", "added"),
|
||||
})
|
||||
|
||||
if diff := schemaNames(reg.Schemas()); !slices.Equal(diff, schemaNames(before)) {
|
||||
t.Fatalf("the provider-visible tools array moved with the catalog: %v, want %v", diff, schemaNames(before))
|
||||
}
|
||||
if !slices.Equal(withdrawn, []string{"mcp__atlas__gone"}) {
|
||||
t.Fatalf("withdrawn = %v, want the dropped pinned tool", withdrawn)
|
||||
}
|
||||
gone, ok := reg.Get("mcp__atlas__gone")
|
||||
if !ok || !tool.IsWithdrawn(gone) {
|
||||
t.Fatalf("the dropped tool is %v (withdrawn=%v), want a withdrawn schema slot", ok, tool.IsWithdrawn(gone))
|
||||
}
|
||||
if _, err := gone.Execute(t.Context(), nil); err == nil || !strings.Contains(err.Error(), `MCP tool "gone" not found on server "atlas"`) {
|
||||
t.Fatalf("calling the withdrawn tool returned %v, want the typed not-found", err)
|
||||
}
|
||||
// What arrived is registered but unpinned: it is reachable through
|
||||
// use_capability, which is the turn tail, not the cached prefix.
|
||||
if _, ok := reg.Get("mcp__atlas__added"); !ok {
|
||||
t.Fatal("the tool the server added never reached the registry")
|
||||
}
|
||||
if reg.ProviderSurfacePinned("mcp__atlas__added") {
|
||||
t.Fatal("a tool that arrived mid-session was pinned into the provider schema")
|
||||
}
|
||||
}
|
||||
|
||||
// A tool the model was never shown has no slot to hold: dropping it is just a
|
||||
// removal, and keeping it would offer a dead call to every later turn.
|
||||
func TestRefreshedMCPToolsDropAnUnpinnedTool(t *testing.T) {
|
||||
reg := tool.NewRegistry()
|
||||
reg.Add(refreshedTool("mcp__atlas__gone", "gone"))
|
||||
reg.SetProviderVisibleTools([]string{"use_capability"})
|
||||
|
||||
if withdrawn := registerRefreshedMCPTools(reg, plugin.Spec{Name: "atlas"}, nil); len(withdrawn) != 0 {
|
||||
t.Fatalf("withdrawn = %v, want nothing announced for a tool the model never saw", withdrawn)
|
||||
}
|
||||
if _, ok := reg.Get("mcp__atlas__gone"); ok {
|
||||
t.Fatal("an unpinned tool the server dropped stayed in the registry")
|
||||
}
|
||||
}
|
||||
|
||||
// A second refresh must not announce the same withdrawal again: the slot is
|
||||
// already held, and the model was already told once.
|
||||
func TestRefreshedMCPToolsAnnounceAWithdrawalOnce(t *testing.T) {
|
||||
reg := tool.NewRegistry()
|
||||
reg.Add(refreshedTool("mcp__atlas__gone", "gone"))
|
||||
reg.SetProviderVisibleTools([]string{"mcp__atlas__gone"})
|
||||
|
||||
first := registerRefreshedMCPTools(reg, plugin.Spec{Name: "atlas"}, nil)
|
||||
second := registerRefreshedMCPTools(reg, plugin.Spec{Name: "atlas"}, nil)
|
||||
if len(first) != 1 || len(second) != 0 {
|
||||
t.Fatalf("announced %v then %v, want the withdrawal stated once", first, second)
|
||||
}
|
||||
if _, ok := reg.Get("mcp__atlas__gone"); !ok {
|
||||
t.Fatal("the withdrawn schema slot was vacated by the second refresh")
|
||||
}
|
||||
}
|
||||
|
||||
// A tool that comes back is a live tool again, not a slot: the model's calls
|
||||
// have to reach the server rather than the message about its absence.
|
||||
func TestRefreshedMCPToolsRestoreAToolThatReturns(t *testing.T) {
|
||||
reg := tool.NewRegistry()
|
||||
reg.Add(refreshedTool("mcp__atlas__gone", "gone"))
|
||||
reg.SetProviderVisibleTools([]string{"mcp__atlas__gone"})
|
||||
|
||||
registerRefreshedMCPTools(reg, plugin.Spec{Name: "atlas"}, nil)
|
||||
registerRefreshedMCPTools(reg, plugin.Spec{Name: "atlas"}, []tool.Tool{refreshedTool("mcp__atlas__gone", "gone")})
|
||||
|
||||
back, ok := reg.Get("mcp__atlas__gone")
|
||||
if !ok || tool.IsWithdrawn(back) {
|
||||
t.Fatalf("the returning tool is %v (withdrawn=%v), want the server's tool back", ok, tool.IsWithdrawn(back))
|
||||
}
|
||||
}
|
||||
|
||||
func schemaNames(schemas []provider.ToolSchema) []string {
|
||||
out := make([]string, 0, len(schemas))
|
||||
for _, s := range schemas {
|
||||
out = append(out, s.Name+"|"+s.Description+"|"+string(s.Parameters))
|
||||
}
|
||||
return out
|
||||
}
|
||||
Reference in new issue
Block a user