diff --git a/pkgs/mcpx/docs/spec/conformance-matrix.md b/pkgs/mcpx/docs/spec/conformance-matrix.md index 8c22df074..23cbb9515 100644 --- a/pkgs/mcpx/docs/spec/conformance-matrix.md +++ b/pkgs/mcpx/docs/spec/conformance-matrix.md @@ -2,7 +2,7 @@ Generated by `internal/conformance` (`MCPX_UPDATE_MATRIX=1 go test ./internal/conformance -run TestMatrix`); do not edit. -Generated from inputs `46df49c7a0e8cc95` (sha256 of docs/spec/requirements.md, internal/conformance/overrides/*.json, internal/conformance/coverage*.go). `TestMatrix` regenerates this file and fails when it differs, so it describes the tree it is committed in. +Generated from inputs `e521a082e8f389bb` (sha256 of docs/spec/requirements.md, internal/conformance/overrides/*.json, internal/conformance/coverage*.go). `TestMatrix` regenerates this file and fails when it differs, so it describes the tree it is committed in. Every normative requirement of every revision (catalogue: `docs/spec/requirements.md`), for mcpx as a server and as a client. A requirement is *tested* when a named test verifies it in that revision, *n/a* when it does not bind mcpx in that role, and a *gap* when it binds mcpx and mcpx does not meet it yet, with the issue that tracks it. Run exactly the covering tests with `go run ./internal/conformance/cmd/covering`. @@ -64,7 +64,7 @@ Every normative requirement of every revision (catalogue: `docs/spec/requirement | 2026-07-28 | server | SHOULD NOT | 11 | 9 | 2 | 0 | 9 | | 2026-07-28 | server | MAY | 31 | 29 | 2 | 0 | 30 | | 2026-07-28 | server | other | 23 | 22 | 1 | 0 | 4 | -| 2026-07-28 | client | MUST | 91 | 62 | 29 | 0 | 133 | +| 2026-07-28 | client | MUST | 91 | 63 | 28 | 0 | 133 | | 2026-07-28 | client | MUST NOT | 26 | 22 | 4 | 0 | 38 | | 2026-07-28 | client | SHOULD | 75 | 52 | 23 | 0 | 85 | | 2026-07-28 | client | SHOULD NOT | 13 | 11 | 2 | 0 | 7 | @@ -2522,7 +2522,7 @@ Every normative requirement of every revision (catalogue: `docs/spec/requirement | [streamable-http-stop-cancelled-work](https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http#cancellation) | SHOULD | server | tested: `conformance.TestTransportServer/2026-07-28/transport-streamable-http/streamable-http-stop-cancelled-work` | n/a: addresses the other side | | [streamable-http-silent-after-cancel](https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http#cancellation) | MUST NOT | server | tested: `conformance.TestTransportServer/2026-07-28/transport-streamable-http/streamable-http-silent-after-cancel` | n/a: addresses the other side | | [streamable-http-no-resumability](https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http#receiving-messages) | — (statement) | both | tested: `conformance.TestTransportServer/2026-07-28/transport-streamable-http/streamable-http-no-resumability` | tested: `conformance.TestTransportClient/2026-07-28/transport-streamable-http/streamable-http-no-resumability` | -| [streamable-http-client-reissue-broken-request](https://modelcontextprotocol.io/specification/2026-07-28/changelog) | MUST (changelog) | client | n/a: addresses the other side | **gap** #203: a broken 2026-07-28 response stream is reported lost to the caller (http.go); nothing re-issues the request with a new id | +| [streamable-http-client-reissue-broken-request](https://modelcontextprotocol.io/specification/2026-07-28/changelog) | MUST (changelog) | client | n/a: addresses the other side | tested: `mcpclient.TestModernReissueBrokenRequest/2026-07-28/transport/client-reissues-broken-request-once` | | [streamable-http-protocol-version-header-every-post](https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http#protocol-version-header) | MUST | client | n/a: addresses the other side | tested: `conformance.TestTransportClient/2026-07-28/transport-streamable-http/streamable-http-protocol-version-header-every-post` | | [streamable-http-protocol-version-header-matches-meta](https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http#protocol-version-header) | MUST | both | tested: `conformance.TestTransportServer/2026-07-28/transport-streamable-http/streamable-http-protocol-version-header-matches-meta` | tested: `conformance.TestTransportClient/2026-07-28/transport-streamable-http/streamable-http-protocol-version-header-matches-meta` | | [streamable-http-unsupported-version-400](https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http#protocol-version-header) | MUST | server | tested: `conformance.TestTransportServer/2026-07-28/transport-streamable-http/streamable-http-unsupported-version-400` | n/a: addresses the other side | diff --git a/pkgs/mcpx/internal/conformance/coverage_transport.go b/pkgs/mcpx/internal/conformance/coverage_transport.go index 69aed3e64..c2264d90e 100644 --- a/pkgs/mcpx/internal/conformance/coverage_transport.go +++ b/pkgs/mcpx/internal/conformance/coverage_transport.go @@ -16,6 +16,7 @@ func init() { C("streamable-http-x-mcp-header-client-must-mirror", "mcpclient.TestParamHeaders/2026-07-28/transport/param-headers-mirror-and-omit-null"), C("streamable-http-client-rejects-invalid-x-mcp-header", "mcpclient.TestToolHeaderAnnotations/2026-07-28/transport/client-rejects-invalid-x-mcp-header", "mcpclient.TestModernHTTPHeaders/2026-07-28/transport/client-excludes-tool-with-invalid-x-mcp-header"), C("streamable-http-client-relist-on-header-mismatch", "mcpclient.TestModernHTTPHeaders/2026-07-28/transport/client-relists-and-retries-on-header-mismatch"), + C("streamable-http-client-reissue-broken-request", "mcpclient.TestModernReissueBrokenRequest/2026-07-28/transport/client-reissues-broken-request-once"), C("streamable-http-client-polls-after-close", "mcpclient.TestLegacyHTTPStreamResumption/2025-11-25/transport/client-resumes-closed-stream-with-last-event-id-after-retry"), C("streamable-http-client-respects-retry", "mcpclient.TestLegacyHTTPStreamResumption/2025-11-25/transport/client-resumes-closed-stream-with-last-event-id-after-retry"), C("streamable-http-client-resume-with-last-event-id", "mcpclient.TestLegacyHTTPStreamResumption/2025-11-25/transport/client-resumes-closed-stream-with-last-event-id-after-retry@2025-03-26,2025-06-18,2025-11-25"), @@ -111,7 +112,6 @@ func init() { ), ) registerGaps( - Gap{ID: "streamable-http-client-reissue-broken-request", Side: Client, Issue: "#203", Why: "a broken 2026-07-28 response stream is reported lost to the caller (http.go); nothing re-issues the request with a new id"}, Gap{ID: "streamable-http-client-reinit-on-404", Side: Client, Issue: "#203", Why: "a 404 for the session is returned as an error; no new initialize"}, Gap{ID: "stdio-client-restart-crashed-server", Side: Client, Issue: "#257", Why: "unverified: the in-flight call fails and the pool respawns only on the next acquire, subject to its start-failure cooldown"}, ) diff --git a/pkgs/mcpx/internal/mcpclient/client.go b/pkgs/mcpx/internal/mcpclient/client.go index 2ad120281..8e9e15d8b 100644 --- a/pkgs/mcpx/internal/mcpclient/client.go +++ b/pkgs/mcpx/internal/mcpclient/client.go @@ -134,6 +134,13 @@ type Client struct { // toolHeaders are each tool's x-mcp-header annotations, learned from // tools/list, for a modern connection over HTTP. toolHeaders map[string][]mcpheaders.Param + // toolHints are each tool's readOnlyHint/idempotentHint annotations, + // learned from tools/list, for the modern broken-stream re-issue + // gate on tools/call: a call may be re-issued only when its tool + // declared itself safe to repeat. Unlike toolHeaders this is not + // gated on mirrorsHeaders -- the gate must work whether or not + // header mirroring is on, and over every transport. + toolHints map[string]toolHint // invalidTools are tools excluded for invalid annotations, with why. invalidTools map[string]string // listen is the subscriptions/listen stream of a modern connection. @@ -216,6 +223,25 @@ type toolsListResult struct { CacheScope string `json:"cacheScope,omitempty"` } +// toolHint is the pair of a tool's behaviour annotations the broken-stream +// re-issue gate reads: a tools/call may be repeated only when the tool +// declared itself read-only or idempotent. +type toolHint struct { + readOnly bool + idempotent bool +} + +// parseToolHint reads the hints out of a tool's raw Annotations. Anything +// unparseable or absent is no hint at all, which closes the gate. +func parseToolHint(raw json.RawMessage) toolHint { + var a struct { + ReadOnly bool `json:"readOnlyHint"` + Idempotent bool `json:"idempotentHint"` + } + _ = json.Unmarshal(raw, &a) + return toolHint{readOnly: a.ReadOnly, idempotent: a.Idempotent} +} + // Resource is one entry from resources/list. type Resource struct { URI string `json:"uri,omitempty"` @@ -937,9 +963,18 @@ func (c *Client) notify(ctx context.Context, method string, params json.RawMessa } func (c *Client) call(ctx context.Context, method string, params json.RawMessage, out any) error { + return c.callReissue(ctx, method, params, out, true) +} + +// callReissue is call with the caller's broken-stream re-issue policy: +// allowReissue says a modern request whose response stream broke +// unanswered may be sent once more with a fresh id. Every method allows +// it except a tools/call whose tool has not declared itself safe to +// repeat (see CallTool). +func (c *Client) callReissue(ctx context.Context, method string, params json.RawMessage, out any, allowReissue bool) error { version := c.metaVersion if version == "" { - raw, err := c.roundTrip(ctx, method, params) + raw, err := c.roundTrip(ctx, method, params, allowReissue) if err != nil || out == nil { return err } @@ -955,7 +990,7 @@ func (c *Client) call(ctx context.Context, method string, params json.RawMessage if err != nil { return err } - raw, err := c.roundTrip(ctx, method, withMeta) + raw, err := c.roundTrip(ctx, method, withMeta, allowReissue) if err != nil { return err } @@ -986,8 +1021,44 @@ func (c *Client) call(ctx context.Context, method string, params json.RawMessage } } +// isStreamLost reports whether err is the synthesized stream-lost error: +// codeStreamLost carrying the mcpxStreamLost data marker lost() stamps on +// it. The code alone does not identify it -- the receive loop's own +// -32000 "connection closed" (see fail) and any server-sent -32000 share +// the code and carry no marker, and neither may be re-issued: transport +// death is a different mechanism, and a server's own error is an answer. +// No message matching anywhere. +func isStreamLost(err error) bool { + var re *rpcError + if !errors.As(err, &re) || re.Code != codeStreamLost { + return false + } + var data struct { + StreamLost bool `json:"mcpxStreamLost"` + } + return json.Unmarshal(re.Data, &data) == nil && data.StreamLost +} + // roundTrip sends one request and returns its raw result. -func (c *Client) roundTrip(ctx context.Context, method string, params json.RawMessage) (json.RawMessage, error) { +// +// On a modern connection a request whose response stream broke +// unanswered (isStreamLost) is re-issued exactly once with a fresh id, +// when allowReissue holds: 2026-07-28 has no resumption, so the broken +// request is the client's to send again or lose. A second stream-lost +// surfaces. Legacy keeps its resume() mechanism in the transport, and +// a caller cancellation is never re-issued. Each input_required round +// is its own call to roundTrip, so each round re-issues independently. +func (c *Client) roundTrip(ctx context.Context, method string, params json.RawMessage, allowReissue bool) (json.RawMessage, error) { + for attempt := 0; ; attempt++ { + raw, err := c.roundTripOnce(ctx, method, params) + if err == nil || attempt > 0 || !allowReissue || c.Era != EraModern || ctx.Err() != nil || !isStreamLost(err) { + return raw, err + } + } +} + +// roundTripOnce sends one request once and returns its raw result. +func (c *Client) roundTripOnce(ctx context.Context, method string, params json.RawMessage) (json.RawMessage, error) { id := c.nextID.Add(1) ch := make(chan *rpcResponse, 1) @@ -1029,6 +1100,12 @@ func (c *Client) roundTrip(ctx context.Context, method string, params json.RawMe return nil, ctx.Err() case resp := <-ch: if resp.Error != nil { + if ctx.Err() != nil && isStreamLost(resp.Error) { + // The caller cancelled while the stream broke: + // cancellation is the truthful outcome, and a + // cancelled request is never re-issued. + return nil, ctx.Err() + } return nil, resp.Error } return resp.Result, nil @@ -1109,10 +1186,25 @@ func (c *Client) ListToolsWithHints(ctx context.Context) ([]Tool, CacheHints, er } cursor = &next } + c.recordToolHints(all) checked := c.checkToolHeaders(c.checkToolSchemas(all)) return checked, hints, nil } +// recordToolHints retains each listed tool's behaviour annotations for +// the tools/call re-issue gate, replacing the previous snapshot whole, +// as checkToolHeaders does for toolHeaders -- but in every era and over +// every transport, since the gate is not a header-mirroring concern. +func (c *Client) recordToolHints(tools []Tool) { + hints := make(map[string]toolHint, len(tools)) + for _, t := range tools { + hints[t.Name] = parseToolHint(t.Annotations) + } + c.mu.Lock() + c.toolHints = hints + c.mu.Unlock() +} + // ListResources returns static resources plus resource templates. func (c *Client) ListResources(ctx context.Context) ([]Resource, error) { all, _, err := c.ListResourcesWithHints(ctx) @@ -1235,8 +1327,29 @@ func (c *Client) GetPrompt(ctx context.Context, name string, args map[string]str return raw, nil } +// toolCallReissueAllowed reports whether a broken tools/call to name may +// be re-issued: only when the tool's retained list-time annotations +// declare readOnlyHint or idempotentHint. It reads only what ListTools +// retained -- a client that has never listed has no retained annotations +// and does not re-issue, and an unknown or unannotated tool closes the +// gate too: re-issuing could execute the tool twice. (Deliberately no +// lazy list here: CallTool must not grow a tools/list round trip on +// servers that may never answer one. toolCallHeaders already lists on +// a modern HTTP connection before the first attempt, so the gate is +// computed after it.) +func (c *Client) toolCallReissueAllowed(name string) bool { + c.mu.Lock() + h, ok := c.toolHints[name] + c.mu.Unlock() + return ok && (h.readOnly || h.idempotent) +} + // CallTool invokes a tool and returns the raw CallToolResult. // +// A broken response stream re-issues the call once only when the tool's +// list-time annotations declare it read-only or idempotent (see +// toolCallReissueAllowed); every other method re-issues automatically. +// // On a modern HTTP connection the tool's x-mcp-header parameters are // mirrored into Mcp-Param-* headers. A -32020 HeaderMismatch then most // likely means the tool's schema changed since it was listed, so the list is @@ -1262,12 +1375,18 @@ func (c *Client) CallTool(ctx context.Context, name string, args any) (json.RawM } defer done() var raw json.RawMessage + allowReissue := false for attempt := 0; ; attempt++ { hctx, err := c.toolCallHeaders(ctx, name, args, attempt > 0) if err != nil { return nil, err } - err = c.call(hctx, "tools/call", params, &raw) + if attempt == 0 { + // After toolCallHeaders: on a modern HTTP connection it + // has just listed, so the retained hints are current. + allowReissue = c.toolCallReissueAllowed(name) + } + err = c.callReissue(hctx, "tools/call", params, &raw, allowReissue) var re *rpcError if attempt == 0 && errors.As(err, &re) && re.Code == codeHeaderMismatch && c.mirrorsHeaders() { continue diff --git a/pkgs/mcpx/internal/mcpclient/http.go b/pkgs/mcpx/internal/mcpclient/http.go index c203d83f4..0aa64602c 100644 --- a/pkgs/mcpx/internal/mcpclient/http.go +++ b/pkgs/mcpx/internal/mcpclient/http.go @@ -378,13 +378,20 @@ func (t *HTTPTransport) resume(ctx context.Context, reqID json.RawMessage, st ss func (t *HTTPTransport) lost(reqID json.RawMessage) { b, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": reqID, "error": map[string]any{ "code": codeStreamLost, "message": "the server closed the response stream before answering", + // The marker is this error's identity: -32000 alone is shared + // with the receive loop's "connection closed" and with anything + // a server sends, and only a stream mcpx itself watched break + // may be re-issued (see isStreamLost). + "data": map[string]any{"mcpxStreamLost": true}, }}) t.push(b) } // codeStreamLost is the JSON-RPC error mcpx reports for a request whose // response stream closed first. -32000 is the start of the range JSON-RPC -// reserves for implementation-defined server errors. +// reserves for implementation-defined server errors. The code alone does +// not identify a lost stream -- the mcpxStreamLost data marker lost() +// stamps on its error does (see isStreamLost). const codeStreamLost = -32000 // get opens a GET stream, resuming after lastID when it is set. diff --git a/pkgs/mcpx/internal/mcpclient/reissue_test.go b/pkgs/mcpx/internal/mcpclient/reissue_test.go new file mode 100644 index 000000000..de5eb0e97 --- /dev/null +++ b/pkgs/mcpx/internal/mcpclient/reissue_test.go @@ -0,0 +1,284 @@ +package mcpclient_test + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + "time" + + "github.com/dezren39/mcpx/internal/mcpclient" +) + +// markedLost is the error frame lost() synthesises when a modern response +// stream breaks unanswered: codeStreamLost with the data marker. +func markedLost() map[string]any { + return rpcErr(-32000, "the server closed the response stream before answering", + map[string]any{"mcpxStreamLost": true}) +} + +type reissueStep struct { + result any + err map[string]any +} + +// reissueScript serves queued steps per key (method, or method:tool for +// tools/call); an exhausted queue answers method-not-found. +type reissueScript struct { + mu sync.Mutex + steps map[string][]reissueStep + hook func(key string) +} + +func (q *reissueScript) serve(method string, params map[string]any) (any, map[string]any) { + key := method + if method == "tools/call" { + if name, _ := params["name"].(string); name != "" { + key = method + ":" + name + } + } + q.mu.Lock() + steps := q.steps[key] + var st reissueStep + if len(steps) == 0 { + st = reissueStep{err: rpcErr(-32601, "no scripted step for "+key, nil)} + } else { + st = steps[0] + q.steps[key] = steps[1:] + } + hook := q.hook + q.mu.Unlock() + if hook != nil { + hook(key) + } + return st.result, st.err +} + +// reissueServer is a modern-only probe server whose non-handshake methods +// are answered by the script. +func reissueServer(q *reissueScript) *probeServer { + return newProbeServer(func(method string, params map[string]any) (any, map[string]any, bool, time.Duration) { + switch method { + case "initialize": + return nil, rpcErr(-32601, "initialize unsupported; supported: 2026-07-28", nil), false, 0 + case "server/discover": + return discoverResult("2026-07-28"), nil, false, 0 + } + r, e := q.serve(method, params) + return r, e, false, 0 + }) +} + +func reissueDial(t *testing.T, s *probeServer) *mcpclient.Client { + t.Helper() + c, err := probeDial(t, s, mcpclient.Options{}) + if err != nil { + t.Fatal(err) + } + if c.Era != mcpclient.EraModern { + t.Fatalf("era = %q, want modern", c.Era) + } + return c +} + +// requestIDs returns the ids of the frames sent for method (and tool, when +// set), in send order. +func requestIDs(s *probeServer, method, tool string) []float64 { + s.mu.Lock() + defer s.mu.Unlock() + var ids []float64 + for _, f := range s.frames { + if f["method"] != method { + continue + } + if tool != "" { + p, _ := f["params"].(map[string]any) + if p["name"] != tool { + continue + } + } + id, _ := f["id"].(float64) + ids = append(ids, id) + } + return ids +} + +func TestModernReissueBrokenRequest(t *testing.T) { + t.Run("2026-07-28", func(t *testing.T) { + t.Run("transport", func(t *testing.T) { + t.Run("client-reissues-broken-request-once", func(t *testing.T) { + q := &reissueScript{steps: map[string][]reissueStep{ + "resources/read": { + {err: markedLost()}, + {result: map[string]any{"contents": []any{}}}, + }, + }} + s := reissueServer(q) + c := reissueDial(t, s) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if _, err := c.ReadResource(ctx, "file:///x"); err != nil { + t.Fatalf("a marked stream-lost must be re-issued and served: %v", err) + } + ids := requestIDs(s, "resources/read", "") + if len(ids) != 2 { + t.Fatalf("resources/read sent %d times, want exactly 2 (original + one re-issue)", len(ids)) + } + if ids[0] == ids[1] { + t.Fatalf("the re-issue must carry a fresh id, both were %v", ids[0]) + } + }) + }) + }) +} + +func TestReissueSecondStreamLostSurfaces(t *testing.T) { + q := &reissueScript{steps: map[string][]reissueStep{ + "resources/read": {{err: markedLost()}, {err: markedLost()}}, + }} + s := reissueServer(q) + c := reissueDial(t, s) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, err := c.ReadResource(ctx, "file:///x") + if err == nil || !strings.Contains(err.Error(), "the server closed the response stream") { + t.Fatalf("a second stream-lost must surface, got %v", err) + } + if ids := requestIDs(s, "resources/read", ""); len(ids) != 2 { + t.Fatalf("resources/read sent %d times, want 2 (one re-issue, no more)", len(ids)) + } +} + +func TestUnmarkedStreamLostErrorsAreNotReissued(t *testing.T) { + cases := map[string]map[string]any{ + // The receive loop's own -32000, as fail() synthesises it. + "connection-closed": rpcErr(-32000, "connection closed: boom", nil), + // A server-sent -32000 with the lost message but no marker: no + // message matching anywhere. + "server-sent-same-message": rpcErr(-32000, "the server closed the response stream before answering", nil), + // A server-sent -32000 whose data lacks the marker. + "server-sent-other-data": rpcErr(-32000, "server error", map[string]any{"other": true}), + } + for name, frame := range cases { + t.Run(name, func(t *testing.T) { + q := &reissueScript{steps: map[string][]reissueStep{ + "resources/read": {{err: frame}, {result: map[string]any{"contents": []any{}}}}, + }} + s := reissueServer(q) + c := reissueDial(t, s) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if _, err := c.ReadResource(ctx, "file:///x"); err == nil { + t.Fatal("an unmarked -32000 must surface, not be re-issued") + } + if ids := requestIDs(s, "resources/read", ""); len(ids) != 1 { + t.Fatalf("resources/read sent %d times, want 1 (no re-issue)", len(ids)) + } + }) + } +} + +func TestReissueSkipsCancelledCaller(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + q := &reissueScript{ + steps: map[string][]reissueStep{ + "resources/read": {{err: markedLost()}, {result: map[string]any{"contents": []any{}}}}, + }, + hook: func(key string) { + if key == "resources/read" { + cancel() + } + }, + } + s := reissueServer(q) + c := reissueDial(t, s) + _, err := c.ReadResource(ctx, "file:///x") + if !errors.Is(err, context.Canceled) { + t.Fatalf("a cancelled caller must see its cancellation, got %v", err) + } + if ids := requestIDs(s, "resources/read", ""); len(ids) != 1 { + t.Fatalf("resources/read sent %d times, want 1 (no re-issue after cancel)", len(ids)) + } +} + +func TestReissueLegacyEraUntouched(t *testing.T) { + q := &reissueScript{steps: map[string][]reissueStep{ + "resources/read": {{err: markedLost()}, {result: map[string]any{"contents": []any{}}}}, + }} + s := newProbeServer(func(method string, params map[string]any) (any, map[string]any, bool, time.Duration) { + switch method { + case "initialize": + return initOK, nil, false, 0 + case "server/discover": + return nil, rpcErr(-32601, "no server/discover", nil), false, 0 + } + r, e := q.serve(method, params) + return r, e, false, 0 + }) + c, err := probeDial(t, s, mcpclient.Options{Preference: mcpclient.ForceLegacy}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if _, err := c.ReadResource(ctx, "file:///x"); err == nil { + t.Fatal("a legacy stream-lost must surface (resume is the legacy mechanism)") + } + if ids := requestIDs(s, "resources/read", ""); len(ids) != 1 { + t.Fatalf("resources/read sent %d times, want 1 (no re-issue in the legacy era)", len(ids)) + } +} + +func TestCallToolReissueAnnotationGate(t *testing.T) { + tool := func(name string, annotations map[string]any) map[string]any { + m := map[string]any{"name": name, "inputSchema": map[string]any{"type": "object"}} + if annotations != nil { + m["annotations"] = annotations + } + return m + } + ok := map[string]any{"content": []any{}} + q := &reissueScript{steps: map[string][]reissueStep{ + "tools/list": {{result: map[string]any{"tools": []any{ + tool("safe", map[string]any{"readOnlyHint": true}), + tool("idem", map[string]any{"idempotentHint": true}), + tool("destructive", map[string]any{"destructiveHint": true}), + tool("plain", nil), + }}}}, + "tools/call:safe": {{err: markedLost()}, {result: ok}}, + "tools/call:idem": {{err: markedLost()}, {result: ok}}, + "tools/call:destructive": {{err: markedLost()}, {result: ok}}, + "tools/call:plain": {{err: markedLost()}, {result: ok}}, + }} + s := reissueServer(q) + c := reissueDial(t, s) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + // The host flow: tools are listed before they are called, which is + // what retains the annotations the gate reads. + tools, err := c.ListTools(ctx) + if err != nil || len(tools) != 4 { + t.Fatalf("ListTools = %d tools, %v; want 4 tools, no error", len(tools), err) + } + + for _, name := range []string{"safe", "idem"} { + if _, err := c.CallTool(ctx, name, map[string]any{}); err != nil { + t.Fatalf("%s: an annotated-safe tool must be re-issued and served: %v", name, err) + } + ids := requestIDs(s, "tools/call", name) + if len(ids) != 2 || ids[0] == ids[1] { + t.Fatalf("%s: tools/call sent with ids %v, want 2 sends with distinct ids", name, ids) + } + } + for _, name := range []string{"destructive", "plain"} { + if _, err := c.CallTool(ctx, name, map[string]any{}); err == nil { + t.Fatalf("%s: an unannotated/non-idempotent tool must not be re-issued", name) + } + if ids := requestIDs(s, "tools/call", name); len(ids) != 1 { + t.Fatalf("%s: tools/call sent %d times, want 1 (gate closed)", name, len(ids)) + } + } +}