Merge pull request #1768 from fahrim/fix/1728-jsonrpc-version-and-reply-id

Fix #1728: require the JSON-RPC 2.0 envelope and correlate control replies

crates/mcp/src/backend.rs

@@ -94,6 +94,23 @@ impl Backend for Remote {
             format!("VectorCraft app at {}: `{method}` failed: {e}; operation may have completed; inspect state before retrying", self.addr)
         })?;
         let v: Value = serde_json::from_str(reply.trim()).map_err(|e| format!("bad reply from app: {e}"))?;
+        if v.get("id").is_none() && v.get("ok").and_then(Value::as_bool) == Some(false) {
+            // The app's refusals carry no id (too many connections, a line it would not read):
+            // it ran nothing and closes the connection, so the next call starts a fresh one.
+            self.conn = None;
+            let why = v.get("error").and_then(Value::as_str).unwrap_or("unknown error");
+            return Err(format!("VectorCraft app at {} refused `{method}`: {why}; nothing was run", self.addr));
+        }
+        if v.get("id") != Some(&json!(id)) {
+            // Not the answer to this call: the stream is desynchronised (or the endpoint faulty),
+            // so the connection is dropped and the next call starts a fresh one. The operation may
+            // still have completed, as with any reply that never arrived whole (#1728).
+            self.conn = None;
+            return Err(format!(
+                "VectorCraft app at {}: reply id mismatch (expected {id}); connection dropped; operation may have completed; inspect state before retrying",
+                self.addr
+            ));
+        }
         if v.get("ok").and_then(Value::as_bool) == Some(true) {
             Ok(v.get("result").cloned().unwrap_or(Value::Null))
         } else {

crates/mcp/src/server.rs

@@ -198,6 +198,12 @@ impl Server {
             }
             return Some(error(id.unwrap_or(Value::Null), INVALID_REQUEST, "missing `method`"));
         };
+        // JSON-RPC 2.0 is the only envelope this server speaks: a request or notification in any
+        // other envelope is never dispatched and never changes state (#1728).
+        if o.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
+            // A notification has no id to address a reply to; ignoring it is the only answer.
+            return id.map(|id| error(id, INVALID_REQUEST, "expected `\"jsonrpc\": \"2.0\"`"));
+        };
         let params = o.get("params").cloned().unwrap_or(Value::Null);
         let Some(id) = id else {
             self.notification(method, &params);

crates/mcp/src/tests.rs

@@ -506,6 +506,71 @@ fn remote_does_not_replay_when_edit_reply_is_lost() {
     assert!(next.is_ok(), "the next call must reconnect: {next:?}");
 }
 
+#[test]
+fn remote_rejects_a_reply_with_a_mismatched_id_and_reconnects() {
+    // A fake control server that answers the first request with another call's id, then serves
+    // correctly: the client must reject the stray reply, drop the connection, and reconnect.
+    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+    let h = std::thread::spawn(move || {
+        let mut seen = vec![];
+        for stream in listener.incoming().flatten().take(2) {
+            stream.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap();
+            let mut out = stream.try_clone().unwrap();
+            let mut reader = BufReader::new(stream);
+            let mut line = String::new();
+            if reader.read_line(&mut line).unwrap_or(0) == 0 {
+                continue;
+            }
+            let msg: Value = serde_json::from_str(&line).unwrap();
+            // First connection: somebody else's answer. Later ones: the right echo.
+            let id = if seen.is_empty() { json!(msg["id"].as_u64().unwrap_or(0) + 1000) } else { msg["id"].clone() };
+            let reply = json!({"ok": true, "result": {"echo": msg["params"]}, "id": id});
+            writeln!(out, "{reply}").unwrap();
+            seen.push(msg);
+        }
+        seen
+    });
+    let mut remote = Remote::connect(&addr).unwrap();
+    let err = remote.call("engine.execute", json!({"command": "shape.rectangle", "params": {}})).expect_err("a stray reply must not be trusted");
+    assert!(err.contains("mismatch"), "{err}");
+    let good = remote.call("engine.execute", json!({"command": "shape.rectangle", "params": {}}));
+    assert!(good.is_ok(), "the next call reconnects: {good:?}");
+    drop(remote);
+    let seen = h.join().unwrap();
+    assert_eq!(seen.len(), 2, "one request per connection: {seen:?}");
+    assert_ne!(seen[0]["id"], seen[1]["id"]);
+}
+
+/// The app's refusals carry no id (a 17th connection gets one and is closed). The call reports the
+/// app's reason, not a mismatch, says nothing ran, and the next call starts a fresh connection.
+#[test]
+fn remote_reports_the_reason_of_a_refusal_without_an_id() {
+    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+    let h = std::thread::spawn(move || {
+        let (mut refused, _) = listener.accept().unwrap();
+        // The request is read first, so the refusal answers it and the close is orderly.
+        refused.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap();
+        let _ = BufReader::new(refused.try_clone().unwrap()).read_line(&mut String::new());
+        writeln!(refused, "{}", json!({"ok": false, "error": "too many control connections"})).unwrap();
+        drop(refused);
+        let (stream, _) = listener.accept().unwrap();
+        stream.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap();
+        let mut out = stream.try_clone().unwrap();
+        let mut line = String::new();
+        BufReader::new(stream).read_line(&mut line).unwrap();
+        let msg: Value = serde_json::from_str(&line).unwrap();
+        writeln!(out, "{}", json!({"ok": true, "result": 7, "id": msg["id"]})).unwrap();
+    });
+    let mut remote = Remote::connect(&addr).unwrap();
+    let err = remote.call("document.inspect", json!({})).expect_err("a refusal is an error");
+    assert!(err.contains("too many control connections") && err.contains("nothing was run"), "{err}");
+    assert!(!err.contains("mismatch") && !err.contains("may have completed"), "{err}");
+    assert_eq!(remote.call("document.inspect", json!({})), Ok(json!(7)), "the next call reconnects");
+    h.join().unwrap();
+}
+
 #[test]
 fn remote_connect_fails_fast() {
     // A port this test holds, bound but not listening, so connecting to it is refused. (A port

crates/mcp/src/tests_protocol.rs

@@ -97,6 +97,50 @@ fn prompt_argument_errors_are_invalid_params() {
     assert_eq!(err_code(&mut s, 3, "prompts/get", json!({})), -32602);
 }
 
+// ---------- envelope ----------
+
+/// Requests and notifications outside a `"jsonrpc": "2.0"` envelope are never dispatched (#1728).
+#[test]
+fn envelopes_without_version_20_are_not_dispatched() {
+    let mut s = server();
+    // A request in no envelope: invalid, and nothing ran.
+    let line = json!({"id": 1, "method": "ping", "params": {}}).to_string();
+    let v: Value = serde_json::from_str(&s.handle_line(&line).unwrap()).unwrap();
+    assert_eq!(v["error"]["code"], -32600, "{v}");
+    assert!(v["error"]["message"].as_str().unwrap_or_default().contains("2.0"), "{v}");
+    // A wrong version is the same.
+    let line = json!({"jsonrpc": "1.0", "id": 2, "method": "ping", "params": {}}).to_string();
+    let v: Value = serde_json::from_str(&s.handle_line(&line).unwrap()).unwrap();
+    assert_eq!(v["error"]["code"], -32600, "{v}");
+    // In a batch the unversioned message is an error entry, not a dispatch; the 2.0 one answers.
+    let batch = json!([
+        {"jsonrpc": "2.0", "id": 3, "method": "ping", "params": {}},
+        {"id": 4, "method": "ping", "params": {}},
+    ])
+    .to_string();
+    let v: Value = serde_json::from_str(&s.handle_line(&batch).unwrap()).unwrap();
+    assert_eq!(v.as_array().map(Vec::len), Some(2), "{v}");
+    assert!(v[0].get("result").is_some(), "{v}");
+    assert_eq!(v[1]["error"]["code"], -32600, "{v}");
+    // The 2.0 request still works.
+    assert_eq!(result(&mut s, 5, "ping", json!({})), json!({}));
+}
+
+#[test]
+fn notification_without_version_20_changes_nothing() {
+    let mut s = server();
+    assert!(!s.is_initialized());
+    // No envelope, no id: there is no address for a reply, so it is ignored — and it must not
+    // flip the server into its initialised state.
+    let line = json!({"method": "notifications/initialized", "params": {}}).to_string();
+    assert!(s.handle_line(&line).is_none());
+    assert!(!s.is_initialized(), "an unversioned notification changed state");
+    // The 2.0 notification still works.
+    let line = json!({"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}).to_string();
+    assert!(s.handle_line(&line).is_none());
+    assert!(s.is_initialized());
+}
+
 // ---------- completions ----------
 
 #[test]

crates/mcp/tests/protocol_props.rs

@@ -216,6 +216,70 @@ fn framing_never_panics_on_malformed_messages() {
     rpc(&mut s, 7, "ping", json!({}));
 }
 
+/// One line of [`limits_and_envelopes_answer_each_line_once`]: `(the line, what must answer it)`.
+/// The limits are written out here (64 messages, 256 steps, 16 MiB): they are the protocol that
+/// `docs/mcp.md` promises, so changing one must fail a test.
+fn limit_line(kind: u8, n: usize, id: u64) -> (String, Value) {
+    let ping = |id: Value| json!({"jsonrpc": "2.0", "id": id, "method": "ping", "params": {}});
+    match kind {
+        0 => (ping(json!(id)).to_string(), json!("result")),
+        // No envelope, or another version: refused when it has an id, ignored when it has none.
+        1 => (json!({"id": id, "method": "ping"}).to_string(), json!(-32600)),
+        2 => (json!({"jsonrpc": "1.0", "id": id, "method": "tools/call", "params": {"name": "undo", "arguments": {}}}).to_string(), json!(-32600)),
+        3 => (json!({"method": "notifications/initialized"}).to_string(), Value::Null),
+        // A batch answers each message, or is refused whole past 64.
+        4 => {
+            let n = 1 + n % 80;
+            (Value::Array((0..n).map(|i| ping(json!(i))).collect()).to_string(), if n > 64 { json!(-32600) } else { json!(n) })
+        }
+        // command_batch refuses more than 256 steps before running any.
+        5 => {
+            let n = 250 + n % 12;
+            let steps: Vec<Value> = (0..n).map(|_| json!({"id": "select.all"})).collect();
+            let call =
+                json!({"jsonrpc": "2.0", "id": id, "method": "tools/call", "params": {"name": "command_batch", "arguments": {"steps": steps}}});
+            (call.to_string(), json!(if n > 256 { "tool error" } else { "result" }))
+        }
+        // A line past 16 MiB is answered and skipped.
+        _ => ("x".repeat(16 * 1024 * 1024 + 1 + n), json!(-32600)),
+    }
+}
+
+proptest! {
+    #![proptest_config(ProptestConfig { cases: 24, ..ProptestConfig::default() })]
+
+    /// The limits and the envelope check (#1727, #1728), from the outside: whatever mix of lines
+    /// arrives, each is answered once (never, for a notification), the server goes on serving
+    /// after every refusal, and no refused line edits the document or initializes the server.
+    #[test]
+    fn limits_and_envelopes_answer_each_line_once(kinds in prop::collection::vec((0u8..6, 0usize..400), 1..12), overlong_at in prop::option::of(0usize..12)) {
+        let mut lines: Vec<(String, Value)> = kinds.iter().enumerate().map(|(i, &(kind, n))| limit_line(kind, n, i as u64)).collect();
+        if let Some(at) = overlong_at {
+            lines.insert(at.min(lines.len()), limit_line(6, at, 0));
+        }
+        let mut s = server();
+        let before = doc_json(&document(&mut s));
+        let input: String = lines.iter().map(|(l, _)| format!("{l}\n")).collect();
+        let mut out = Vec::new();
+        s.serve(input.as_bytes(), &mut out).unwrap();
+        let replies: Vec<Value> = String::from_utf8(out).unwrap().lines().map(|l| serde_json::from_str(l).unwrap()).collect();
+        let expected: Vec<&Value> = lines.iter().map(|(_, want)| want).filter(|w| !w.is_null()).collect();
+        prop_assert_eq!(replies.len(), expected.len(), "one reply per answered line: {:?}", &expected);
+        for (reply, want) in replies.iter().zip(expected) {
+            match want {
+                Value::String(kind) => {
+                    prop_assert_eq!(reply["result"]["isError"].as_bool().unwrap_or(false), kind == "tool error", "{}", reply);
+                    prop_assert!(reply.get("error").is_none(), "{}", reply);
+                }
+                Value::Number(n) if n.as_i64() == Some(-32600) => prop_assert_eq!(&reply["error"]["code"], want, "{}", reply),
+                _ => prop_assert_eq!(reply.as_array().map(Vec::len), want.as_u64().map(|n| n as usize), "{}", reply),
+            }
+        }
+        prop_assert!(!s.is_initialized(), "a notification without the envelope initialized the server");
+        prop_assert_eq!(doc_json(&document(&mut s)), before, "a refused or read-only line changed the document");
+    }
+}
+
 #[test]
 fn resources_are_json() {
     let mut s = server();

docs/mcp.md

@@ -105,7 +105,13 @@ given rather than guarding against other software on the computer.
 ## Protocol
 
 Newline-delimited JSON-RPC 2.0 on stdio. The revision is **`2025-06-18`**; `2025-03-26` and `2024-11-05` are
-accepted too. Clients requesting `2026-07-28` through `initialize`, or through a request's
+accepted too. Every request and notification must carry `"jsonrpc": "2.0"`: anything else is
+refused with `-32600` (requests) or ignored (notifications, which have no id to address a reply
+to) before it is dispatched. When a tool call reaches a running app, it is one control-channel
+request, and the app echoes its `id` in the reply: a reply with any other `id` is rejected, the
+connection dropped, and the next call starts a fresh one — the call reports uncertainty like any
+reply that never arrived whole. The app's refusals carry no `id` (a connection past its 16, a line it
+would not read): the call reports the app's reason and says nothing was run. Clients requesting `2026-07-28` through `initialize`, or through a request's
 `_meta["io.modelcontextprotocol/protocolVersion"]`, receive `resultType: "complete"`,
 `ttlMs` and `cacheScope: "private"` on tool/resource/template/prompt lists and resource/prompt reads.
 List TTL is 600000 ms; live reads use zero. Older clients keep the original response shapes.