mcpx: re-issue a broken modern request once, tools/call gated on annotations (#1638)

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 |

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"},
 	)

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

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.