fix(transport): preserve protocol error replies across cancellation

[rksharma-owg]

Preserve AsyncRwTransport's Invalid Request reply when receive() is cancelled while waiting for the writer or flushing a partial response.

Motivation and Context

Closes #1353.

The service loop polls receive() inside tokio::select!. After consuming a well-formed JSON line with an invalid JSON-RPC shape, receive() awaited the writer lock and send() without retaining reply progress. Cancellation could therefore lose the reply or leave a truncated JSON frame.

Keep queue/flush state on the transport, enqueue the error once with feed(), and resume flushing before reading the next message. The change is confined to async_rw.rs and adds no dependencies or background tasks.

How Has This Been Tested?

Two deterministic regressions use real Tokio duplex streams: one holds the writer lock; the other limits output capacity to one byte. Each cancels receive(), resumes it, and checks one complete -32600 reply plus delivery of the following valid request. Both fail before the fix and pass after it.

Local validation with Rust 1.96.0:

• cargo test --all-features
• cargo test -p rmcp --features "$FEATURES" with all non-internal features except local
• cargo test -p rmcp --lib --no-default-features
• Strict Clippy with all features and with non-local features; nightly formatting check
• Focused transport suite: 12 passed

Fork preflight: all 19 jobs passed against b427e9771a24e9367cf069340631cd9451f97905, based on 5d9935d0fc61affcfc573d475ca0a20e125467be. This includes upstream feature linting, tests, coverage, MSRV, SemVer/public API checks, examples/WASI, documentation, dependency audit, conformance, and CodeQL. Existing ignored tests and documented extension expected failures remain unchanged. The validation workflows are fork-only and absent from this diff. A supplemental fork Scorecard run also passed; its pinned action requires a default-branch run, so it was run manually without publishing Scorecard results.

Breaking Changes

None identified. Public APIs and the wire format are unchanged; syntactically invalid JSON is still ignored as before.

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)
• Documentation update

Checklist

• I have read the MCP Documentation
• My code follows the repository's style guidelines
• New and existing tests pass locally
• I have added appropriate error handling
• I have added or updated documentation as needed (N/A: restores the existing protocol-error behavior)

Additional context

The regressions also reject duplicate replies. Existing input-length limits and close/write-error behavior are preserved.

[github-actions]

@connor4312 (assignment failed) is on triage rotation this week and is the reviewer for this one.

Labels come from the PR labeler. Please review the change.

_{Automated triage. Labels are a first pass and the assigned maintainer has the final say.}

[connor4312]

_{Connor's agent driving:}

Thanks, this looks good! The new tests fail on main and pass with this change.

One small thing for later (not blocking): close() drops the writer without flushing it. So if the server shuts down while this error reply is only partly written, the client gets a cut-off line. I checked this by stopping receive() partway through the reply with a 1-byte duplex and then calling close(): the client only got {. In the service loop, this happens when serve_loop_ct.cancelled() wins the select! and the drain step has no other messages to send.

This problem was already there before this PR. A best-effort flush in close() would fix it, ideally with a time limit so a client that isn't reading can't block shutdown. That could be a follow-up issue.

[DaleSeo]

Thanks for the fix, @rksharma-owg. I only have a non-blocking comment.

[DaleSeo]

nit: With the field and enum documented, this comment repeats what they say and can go.

Suggested change

	// Keep reply progress across cancellation at either the writer lock or
	// flush. Once queued, resuming must flush rather than enqueue a duplicate.