From 80b2bbecfb00433c6c48cb3cf7488c9af86017b3 Mon Sep 17 00:00:00 2001 From: Pranit Date: Fri, 23 Oct 2026 12:10:32 -0400 Subject: [PATCH] Explain oversized MCP replies and stop fetching them again - When a stdio MCP server replies with more than 16 MiB, the model is told the reply was too large and fx restarted the server, instead of "the server went away". - When an HTTP MCP server replies with more than 16 MiB, fx ends the call at once and tells the model the reply was too large, instead of "the server went away". - fx no longer resumes an HTTP stream (2025 protocol) to fetch an oversized reply again. Before, it downloaded the same reply up to two more times. - fx no longer requests an oversized tool list again from an HTTP server on the 2026 protocol. - The debug trace logs each dropped reply with the server, transport, and reason. --- src/core/mcp_host/runtime.zig | 10 +++++ src/mcp/core/http_legacy.zig | 22 ++++++++++ src/mcp/core/http_modern.zig | 28 ++++++++++++- src/mcp/engine.zig | 4 +- src/mcp/io/http.zig | 53 +++++++++++++++++++++++-- src/mcp/io/process.zig | 44 ++++++++++++++++++-- src/mcp/server.zig | 30 +++++++++++--- tests/e2e/fixtures/mcp-legacy-remote.ts | 27 +++++++++++++ tests/e2e/mcp-legacy-remote.test.ts | 21 ++++++++++ tests/e2e/mcp-stdio.test.ts | 34 ++++++++++++++++ tests/tla/mcp/HttpLegacy.cfg | 1 + tests/tla/mcp/HttpLegacy.mutants.json | 6 +++ tests/tla/mcp/HttpLegacy.stress.cfg | 1 + tests/tla/mcp/HttpLegacy.tla | 19 +++++++-- tests/tla/mcp/HttpLegacyTrace.tla | 1 + tests/tla/mcp/HttpModern.cfg | 1 + tests/tla/mcp/HttpModern.mutants.json | 6 +++ tests/tla/mcp/HttpModern.stress.cfg | 1 + tests/tla/mcp/HttpModern.tla | 16 ++++++++ tests/tla/mcp/HttpModernTrace.tla | 1 + tests/tla/mcp/rules.md | 2 + 21 files changed, 311 insertions(+), 17 deletions(-) diff --git a/src/core/mcp_host/runtime.zig b/src/core/mcp_host/runtime.zig index bc2e5d7a6..94b046f49 100644 --- a/src/core/mcp_host/runtime.zig +++ b/src/core/mcp_host/runtime.zig @@ -1067,6 +1067,16 @@ pub const Runtime = struct { return r.finish(req, .{ .failed = err }); }; if (req.declined and result == .ended and result.ended.outcome == .cancelled) result.ended.outcome = .unsupported_input; + // fx dropped the reply, and a stdio server with it: the + // server didn't go away by itself. + if (c.oversized and result == .ended) { + const s = r.resolved.servers[i]; + debug_trace.logf("mcp", "MCP call reply dropped server={s} transport={s} reason=too_large", .{ s.name, @tagName(s.transport) }); + result.ended.why = if (s.transport == .stdio) + "its reply was larger than fx reads in one message, so fx restarted the server" + else + "its reply was larger than fx reads in one message, so fx dropped it"; + } if (result == .ended and result.ended.outcome == .transport_failed) { const down = if (r.enginePhase(i)) |p| p == .backoff or p == .failed else false; if (down) result.ended.why = r.failureText(req.alloc, i) catch null; diff --git a/src/mcp/core/http_legacy.zig b/src/mcp/core/http_legacy.zig index fa3b6af2a..c437dfe87 100644 --- a/src/mcp/core/http_legacy.zig +++ b/src/mcp/core/http_legacy.zig @@ -60,6 +60,8 @@ pub const Event = union(enum) { answered: Id, /// The stream broke before the answer. broke: Id, + /// The answer was over the message limit (LHTTP-11). + too_large: Id, /// A status without a JSON-RPC answer, other than a session 404. refused: Id, /// A 404 to a POST or resume that carried a session id. @@ -238,6 +240,13 @@ pub const Session = struct { } else s.end(r, .lost, out); s.record("break", id, s.phase, out); }, + // Resuming would only bring the same answer: lost at once. + .too_large => |id| { + const r = s.find(id) orelse return error.UnknownRequest; + if (r.state != .posted and r.state != .resuming) return s.ignore("too_large", id, out); + s.end(r, .lost, out); + s.record("too_large", id, s.phase, out); + }, .refused => |id| { const r = s.find(id) orelse return error.UnknownRequest; if (r.state != .posted and r.state != .resuming) return s.ignore("refused", id, out); @@ -526,6 +535,7 @@ fn candidates(s: *Session, w: *const World, ids: []const Id, buf: []Event) []Eve if (w.events < 3) add(buf, &n, .{ .event_seen = .{ .id = id, .has_id = true, .retry_ms = 5 } }); add(buf, &n, .{ .answered = id }); add(buf, &n, .{ .broke = id }); + add(buf, &n, .{ .too_large = id }); add(buf, &n, .{ .refused = id }); if (r.session != 0 and w.server_ended[r.session]) add(buf, &n, .{ .expired = id }); }, @@ -628,6 +638,18 @@ test "a broken stream resumes after its event id and retry, and is lost without try testing.expectEqualSlices(Effect, &.{.{ .deliver = .{ .id = 2, .outcome = .lost } }}, out.effects()); } +test "an answer over the message limit is lost at once, even on a stream that could resume" { + var slots: [2]Request = undefined; + var s: Session = .init(&slots, .{ .listen = false }); + _ = try steps(&s, &.{ .start, .{ .initialize_answered = .{ .session = true } }, .{ .send = 1 }, .{ .send = 2 } }); + var out = try steps(&s, &.{ .{ .event_seen = .{ .id = 1, .has_id = true, .retry_ms = 500 } }, .{ .too_large = 1 } }); + try testing.expectEqualSlices(Effect, &.{.{ .deliver = .{ .id = 1, .outcome = .lost } }}, out.effects()); + // On a resumed stream too. + _ = try steps(&s, &.{ .{ .event_seen = .{ .id = 2, .has_id = true, .retry_ms = 5 } }, .{ .broke = 2 }, .{ .resume_due = 2 } }); + out = try steps(&s, &.{.{ .too_large = 2 }}); + try testing.expectEqualSlices(Effect, &.{.{ .deliver = .{ .id = 2, .outcome = .lost } }}, out.effects()); +} + test "a stream isn't resumed in a newer session" { var slots: [2]Request = undefined; var s: Session = .init(&slots, .{ .listen = false }); diff --git a/src/mcp/core/http_modern.zig b/src/mcp/core/http_modern.zig index b01e950b5..91bb54aa1 100644 --- a/src/mcp/core/http_modern.zig +++ b/src/mcp/core/http_modern.zig @@ -30,6 +30,8 @@ pub const Event = union(enum) { http_failure, /// The connection or stream broke before the answer. stream_broken, + /// The answer was over the message limit (HTTP-14). + too_large, /// The re-list after -32020 ended, well or not. relisted: bool, /// The host cancelled, or the request's timeout fired. @@ -128,6 +130,13 @@ pub const Exchange = struct { .http_failure => if (x.phase == .posted) x.finish(.failed, .http_failure, out) else { ignored = true; }, + // Another attempt would only bring the same answer: lost at once. + .too_large => if (x.phase != .posted) { + ignored = true; + } else { + x.maybe_ran = x.kind == .call; + x.finish(.lost, .lost, out); + }, .stream_broken => if (x.phase != .posted) { ignored = true; } else if (x.kind == .idempotent and x.attempts < x.config.max_attempts) { @@ -233,7 +242,7 @@ const World = struct { const moves = [_]Event{ .send, .{ .answer = .result }, .{ .answer = .failure }, .{ .answer = .mismatch }, .http_failure, .stream_broken, .{ .relisted = true }, .{ .relisted = false }, - .cancel, + .cancel, .too_large, }; fn apply(x: *Exchange, w: *World, event: Event) !void { @@ -246,7 +255,7 @@ fn apply(x: *Exchange, w: *World, event: Event) !void { try testing.expectEqual(@as(usize, 0), out.effect_count); return; } - if (event == .stream_broken or event == .answer or event == .http_failure) w.posted = false; + if (event == .stream_broken or event == .too_large or event == .answer or event == .http_failure) w.posted = false; if (event == .relisted) w.relisting = false; for (out.effects()) |effect| switch (effect) { .post => { @@ -285,6 +294,8 @@ fn apply(x: *Exchange, w: *World, event: Event) !void { if (event == .answer) try testing.expect(w.closed); // Where the model has no choice, the core must act. if (event == .stream_broken and before.phase == .posted and x.kind == .call) try testing.expectEqual(Phase.lost, x.phase); + // TooLargeEnds: an answer over the limit is never asked for again. + if (event == .too_large and before.phase == .posted) try testing.expectEqual(Phase.lost, x.phase); if (event == .cancel) try testing.expect(x.phase == .cancelled); if (event == .relisted and !event.relisted) try testing.expectEqual(Phase.done, x.phase); } @@ -348,6 +359,19 @@ test "a listing is re-issued once after a broken stream; a call is lost, maybe r try testing.expectEqualSlices(Effect, &.{.{ .deliver = .{ .outcome = .lost, .maybe_ran = true } }}, out.effects()); } +test "an answer over the message limit is lost at once: a listing isn't re-issued, and a call may have run" { + var list: Exchange = .init(.{}, .idempotent); + var out: Output = .{}; + list.step(.send, &out); + list.step(.too_large, &out); + try testing.expectEqualSlices(Effect, &.{.{ .deliver = .{ .outcome = .lost, .maybe_ran = false } }}, out.effects()); + + var call: Exchange = .init(.{}, .call); + call.step(.send, &out); + call.step(.too_large, &out); + try testing.expectEqualSlices(Effect, &.{.{ .deliver = .{ .outcome = .lost, .maybe_ran = true } }}, out.effects()); +} + test "-32020 to a call re-lists, then retries once" { var call: Exchange = .init(.{}, .call); var out: Output = .{}; diff --git a/src/mcp/engine.zig b/src/mcp/engine.zig index ad4b6dbc9..2c0e3fe90 100644 --- a/src/mcp/engine.zig +++ b/src/mcp/engine.zig @@ -61,7 +61,8 @@ pub const Event = struct { phase: Phase, /// A call ended. `answer` is set when the server answered, valid /// until the next `next`. - called: struct { call: Handle, outcome: Outcome, maybe_ran: bool, timed_out: bool, answer: ?tools.Call }, + /// `oversized`: a lost call's server was killed for a stdout line over the limit. + called: struct { call: Handle, outcome: Outcome, maybe_ran: bool, timed_out: bool, answer: ?tools.Call, oversized: bool = false }, /// A call's server reported progress; `message` is terminal-safe and /// valid until the next `next`. progress: struct { call: Handle, progress: f64, total: ?f64, message: ?[]const u8 }, @@ -446,6 +447,7 @@ pub const Engine = struct { .maybe_ran = c.maybe_ran, .timed_out = c.timed_out, .answer = c.answer, + .oversized = c.oversized, } }); }, .input => |q| for (slot.calls.items) |x| if (x.sent != null and x.sent.? == q.handle) { diff --git a/src/mcp/io/http.zig b/src/mcp/io/http.zig index b37a04bc6..f86ff4267 100644 --- a/src/mcp/io/http.zig +++ b/src/mcp/io/http.zig @@ -37,8 +37,11 @@ pub const Ending = enum { accepted, /// A status with no JSON-RPC message in the body. http_failure, - /// The connection or stream broke, or a message was over the limit. + /// The connection or stream broke. broken, + /// A message was over `max_message`, so the rest of the stream was left + /// unread. Asking for it again would bring the same message. + too_large, /// 401, or 403 with a `WWW-Authenticate` challenge: the server /// didn't take the request. rejected, @@ -459,7 +462,8 @@ pub const Client = struct { } c.readBody(reader, &read, &answer) catch |err| switch (err) { error.Canceled => return error.Canceled, - error.TooLong, error.ReadFailed, error.OutOfMemory => return .{ .status = status, .how = .broken }, + error.TooLong => return .{ .status = status, .how = .too_large }, + error.ReadFailed, error.OutOfMemory => return .{ .status = status, .how = .broken }, }; if (wire.decode(answer.items)) |message| { // A result says it worked, so one under an error status @@ -589,7 +593,7 @@ pub const Client = struct { error.Canceled => return error.Canceled, error.ReadFailed, error.OutOfMemory => return .{ .status = status, .how = .broken }, }; - if (got.too_long) return .{ .status = status, .how = .broken }; + if (got.too_long) return .{ .status = status, .how = .too_large }; const event = parser.line(c.gpa, &data, mark) catch return .{ .status = status, .how = .broken }; if (event) |e| { const event_id: ?[]u8 = if (e.id) |i| c.gpa.dupe(u8, i) catch return .{ .status = status, .how = .broken } else null; @@ -709,6 +713,49 @@ fn serveEvents(server: *Io.net.Server, count: usize) !void { try w.interface.flush(); } +/// Answers one request on `server` with `body` as `content_type`, and closes. +fn serveBody(server: *Io.net.Server, content_type: []const u8, body: []const u8) !void { + const stream = try server.accept(testing.io); + defer stream.close(testing.io); + var in: [4096]u8 = undefined; + var r = stream.reader(testing.io, &in); + while ((try r.interface.takeDelimiterInclusive('\n')).len > 2) {} + var out: [1024]u8 = undefined; + var w = stream.writer(testing.io, &out); + try w.interface.print("HTTP/1.1 200 OK\r\ncontent-type: {s}\r\nconnection: close\r\n\r\n", .{content_type}); + try w.interface.writeAll(body); + try w.interface.flush(); +} + +test "an answer over max_message ends too_large, not broken, in a JSON body and in an SSE event" { + const pad: [200]u8 = @splat('x'); + const json = "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"t\":\"" ++ pad ++ "\"}}"; + const Case = struct { content_type: []const u8, body: []const u8 }; + for ([_]Case{ + .{ .content_type = "application/json", .body = json }, + .{ .content_type = "text/event-stream", .body = "id: e1\ndata: " ++ json ++ "\n\n" }, + }) |case| { + var address: Io.net.IpAddress = .{ .ip4 = .loopback(0) }; + var server = try address.listen(testing.io, .{ .reuse_address = true }); + defer server.deinit(testing.io); + var serving = try testing.io.concurrent(serveBody, .{ &server, case.content_type, case.body }); + defer serving.cancel(testing.io) catch {}; + var url: [64]u8 = undefined; + var c: Client = undefined; + try c.init(testing.io, testing.allocator, .{ .url = try std.mem.print(&url, "http://127.0.0.1:{d}/mcp", .{server.socket.address.getPort()}), .max_message = 64 }); + defer c.deinit(); + try c.send(1, .post, "{}", &.{}); + while (true) switch (try c.next(5_000)) { + .ended => |e| { + try testing.expectEqual(Ending.too_large, e.how); + break; + }, + .timeout => return error.TestTimeout, + else => {}, + }; + } +} + test "events past the budget all arrive, each once the one before it is taken" { var address: Io.net.IpAddress = .{ .ip4 = .loopback(0) }; var server = try address.listen(testing.io, .{ .reuse_address = true }); diff --git a/src/mcp/io/process.zig b/src/mcp/io/process.zig index 55e913d3b..077f68384 100644 --- a/src/mcp/io/process.zig +++ b/src/mcp/io/process.zig @@ -153,7 +153,8 @@ pub const Incoming = union(enum) { dropped: struct { err: wire.DecodeError, line: []const u8 }, /// The connection changed state. `lost` means the process is gone and /// requests in flight on it are lost (`transport_lost` in `core/request.zig`). - changed: struct { state: core.State, lost: bool }, + /// `oversized` says fx killed it for a stdout line over `max_line`. + changed: struct { state: core.State, lost: bool, oversized: bool = false }, /// `next` was given a timeout and it passed first. timeout, }; @@ -271,6 +272,9 @@ pub const Process = struct { budget: Budget, /// Why the last start failed, for logs. spawn_error: ?SpawnError = null, + /// The current process is being killed because it wrote a stdout line + /// over `max_line`: the loss its exit reports is said to be that. + killed_oversized: bool = false, /// How the last process ended, once it has: set under `pid_lock` before /// `.exited` is posted. A status can be read before the owner takes that /// event, so read it with `exitStatus`. @@ -404,13 +408,19 @@ pub const Process = struct { p.held = bytes; return .{ .stderr = bytes }; }, - .oversized => |g| (try p.apply(.{ .oversized_line = g })).lost, + .oversized => |g| blk: { + const applied = try p.apply(.{ .oversized_line = g }); + if (before == .running and p.conn.state == .killing) p.killed_oversized = true; + break :blk applied.lost; + }, .exited => |g| (try p.apply(.{ .process_exited = g })).lost, .timer => |t| (try p.apply(.{ .timer_fired = t })).lost, // One from an earlier process is stale. .write_failed => |g| if (g == p.conn.generation) (try p.apply(.write_failed)).lost else false, }; - if (lost or p.conn.state != before) return .{ .changed = .{ .state = p.conn.state, .lost = lost } }; + const oversized = lost and p.killed_oversized; + if (lost) p.killed_oversized = false; + if (lost or p.conn.state != before) return .{ .changed = .{ .state = p.conn.state, .lost = lost, .oversized = oversized } }; return null; } @@ -436,6 +446,7 @@ pub const Process = struct { for (out.effects()) |effect| switch (effect) { .spawn => |g| { p.spawn_error = null; + p.killed_oversized = false; p.pid_lock.lockUncancelable(p.io); p.exit = null; p.pid_lock.unlock(p.io); @@ -980,6 +991,33 @@ test "an oversized line kills the server" { try testing.expect(p.conn.errored and !p.conn.term_sent); } +test "the loss from killing a server for an oversized line says so, and an ordinary exit's doesn't" { + const big = shell("printf '%0100d\\n' 0; exec cat >/dev/null"); + var p: Process = undefined; + initTest(&p, &big); + defer p.deinit(); + try p.start(); + const lost = for (0..200) |_| switch (try p.next(5_000)) { + .changed => |c| if (c.lost) break c, + .timeout => return error.TestTimeout, + else => {}, + } else return error.TestUnexpectedResult; + try testing.expect(lost.oversized); + try testing.expect(!p.killed_oversized); + + const quits = shell("exit 3"); + var q: Process = undefined; + initTest(&q, &quits); + defer q.deinit(); + try q.start(); + const gone = for (0..200) |_| switch (try q.next(5_000)) { + .changed => |c| if (c.lost) break c, + .timeout => return error.TestTimeout, + else => {}, + } else return error.TestUnexpectedResult; + try testing.expect(!gone.oversized); +} + test "stderr lines reach the host and never change the state" { const argv = shell("echo warming up >&2; exec cat >/dev/null"); var p: Process = undefined; diff --git a/src/mcp/server.zig b/src/mcp/server.zig index e24fb696d..a189dc2dd 100644 --- a/src/mcp/server.zig +++ b/src/mcp/server.zig @@ -125,7 +125,8 @@ pub const Event = union(enum) { /// that keeps its own copy lists again before its next call (CAT-05). tools_stale, /// A call ended. `answer` is set when the server answered, valid until the next `next`. - called: struct { handle: Handle, outcome: CallOutcome, maybe_ran: bool, timed_out: bool, answer: ?tools.Call }, + /// `oversized` says a lost call's server was killed for a stdout line over the limit. + called: struct { handle: Handle, outcome: CallOutcome, maybe_ran: bool, timed_out: bool, answer: ?tools.Call, oversized: bool = false }, /// A call's server reported progress that the request core accepted. /// `message` is terminal-safe and cut to 160 bytes, valid until the /// next `next`. @@ -273,6 +274,10 @@ pub const Server = struct { event_count: usize = 0, /// The answer an event points into, kept until the next `next`. held_answer: ?tools.Call = null, + /// Set while requests end lost because an answer was over the message + /// limit: a stdio process fx killed for an oversized line, or an HTTP + /// answer over `max_message`. + losing_oversized: bool = false, /// The message of an invalid reply's answer, kept as long as the line it /// came from: until the process is read again. refused_reply: [512]u8 = undefined, @@ -755,7 +760,7 @@ pub const Server = struct { // With no sink for stderr. .stderr => {}, .dropped => |d| try r.onRefusedLine(d.err, d.line), - .changed => |c| try r.onProcessChanged(c.state, c.lost), + .changed => |c| try r.onProcessChanged(c.state, c.lost, c.oversized), .timeout => unreachable, } } @@ -783,10 +788,12 @@ pub const Server = struct { try r.onMessage(id, 200, .{ .response = .{ .id = id, .body = .{ .failure = .{ .code = wire.code.invalid_request, .message = w.buffered(), .data = null } } } }); } - fn onProcessChanged(r: *Server, state: stdio_conn.State, lost: bool) Error!void { + fn onProcessChanged(r: *Server, state: stdio_conn.State, lost: bool, oversized: bool) Error!void { if (lost) { // What was in flight on the process is lost. The exit ends every // listen, and its elicitations are over. + r.losing_oversized = oversized; + defer r.losing_oversized = false; for (r.table_slots) |slot| if (slot.used) try r.stepRequest(.{ .transport_lost = slot.id }); if (r.subs.up) try r.stepSubs(.transport_down); try r.endElicitations(); @@ -1240,6 +1247,12 @@ pub const Server = struct { if (l.what == .discover) try r.stepEra(era.classifyDiscoverHttp(r.detector.config, status, null)); return r.stepExchange(l, .http_failure); } + // Another attempt would bring the same answer again (HTTP-14). + if (how == .too_large) { + r.losing_oversized = true; + defer r.losing_oversized = false; + return r.stepExchange(l, .too_large); + } try r.stepExchange(l, .stream_broken); } @@ -1330,6 +1343,7 @@ pub const Server = struct { .maybe_ran = o == .lost, .timed_out = l.timed_out, .answer = if (o == .result or o == .failure) r.held_answer else null, + .oversized = o == .lost and r.losing_oversized, }, }); l.used = false; @@ -1636,7 +1650,7 @@ pub const Server = struct { const s = &r.legacy.?; const req = s.find(key) orelse return; if (req.state != .posted and req.state != .resuming) return; - try r.stepLegacy(if (status == 404) + const event: http_legacy.Event = if (status == 404) // `expired` takes a request without a session id for a refusal. .{ .expired = key } else if (status >= 300) @@ -1645,8 +1659,13 @@ pub const Server = struct { // A stream that ended or broke before the answer, which the // server may have closed to be polled. .complete, .accepted, .broken => .{ .broke = key }, + // Resuming would bring the same answer again (LHTTP-11). + .too_large => .{ .too_large = key }, .http_failure, .rejected => .{ .refused = key }, - }); + }; + r.losing_oversized = event == .too_large; + defer r.losing_oversized = false; + try r.stepLegacy(event); } /// Any refusal of the listening GET but a session 404 means the @@ -1688,6 +1707,7 @@ pub const Server = struct { .maybe_ran = outcome == .lost, .timed_out = l.timed_out, .answer = if (outcome == .result) r.held_answer else null, + .oversized = outcome == .lost and r.losing_oversized, }, }); if (l.what == .list and outcome != .result) { diff --git a/tests/e2e/fixtures/mcp-legacy-remote.ts b/tests/e2e/fixtures/mcp-legacy-remote.ts index a3539ee3e..0848b52f9 100644 --- a/tests/e2e/fixtures/mcp-legacy-remote.ts +++ b/tests/e2e/fixtures/mcp-legacy-remote.ts @@ -26,6 +26,7 @@ type StreamableMode = | "mixed_delimiters" | "unsafe_resume_id" | "resume" + | "oversized_resumable" | "poll" | "hold_open_call" | "hold_initialize" @@ -248,6 +249,11 @@ export function startLegacyStreamableHttpFixture( { headers: { "content-type": "text/event-stream" } }, ); } + // Resuming the oversized answer's stream gets the same answer again. + if (mode === "oversized_resumable") { + if (headers["last-event-id"] === undefined) return new Response(null, { status: 405 }); + return sseResponse([{ id: "call-event-2", data: oversizedCallResponse(resumableCallId!) }]); + } return sseResponse([ { id: `resume-${resumeCalls}`, @@ -583,6 +589,22 @@ export function startLegacyStreamableHttpFixture( }, ]); } + // An event id first, so the stream could be resumed, then an answer + // over fx's 16 MiB message limit. + if (mode === "oversized_resumable") { + resumableCallId = message!.id; + return sseResponse([ + { + id: "call-event-1", + data: { + jsonrpc: "2.0", + method: "notifications/progress", + params: { progressToken: message!.id, progress: 1, total: 2 }, + }, + }, + { id: "call-event-2", data: oversizedCallResponse(message!.id!) }, + ]); + } if (mode === "poll") { return new Response("id:\ndata:\n\n", { headers: { "content-type": "text/event-stream" }, @@ -947,6 +969,11 @@ function toolCallResponse(id: number, text: string) { }; } +/** A tools/call answer over fx's 16 MiB message limit. */ +function oversizedCallResponse(id: number) { + return toolCallResponse(id, "x".repeat(17 * 1024 * 1024)); +} + function sseResponse( events: Array<{ id?: string; data: unknown }>, keepOpen = false, diff --git a/tests/e2e/mcp-legacy-remote.test.ts b/tests/e2e/mcp-legacy-remote.test.ts index 5325a99b0..a2508c39d 100644 --- a/tests/e2e/mcp-legacy-remote.test.ts +++ b/tests/e2e/mcp-legacy-remote.test.ts @@ -326,6 +326,27 @@ describe("version-scoped legacy MCP remote transports", () => { }, 30_000); } + // fx reads at most 16 MiB of one message. Before, an answer over that was + // treated as a broken stream: fx resumed it twice, getting the same answer + // each time, then told the model the server went away. + test("Streamable HTTP ends a call whose answer is over fx's message limit at once, without resuming it", async () => { + streamable = startLegacyStreamableHttpFixture("2025-06-18", { mode: "oversized_resumable" }); + const root = createRoot("oversized-answer", "http", streamable.url); + gateway = startToolGateway("Oversized answer handled."); + + const result = await runAsk(root, gateway, "Call the oversized fixture."); + + expect(result.code).toBe(0); + // What the model was told: the call's error text. + const told = /MCP server 'fixture' ended the call[^"\\]*/.exec(gateway.requests[2]?.body ?? "")?.[0] ?? ""; + expect(told).toContain("its reply was larger than fx reads in one message, so fx dropped it"); + expect(told).not.toContain("the server went away"); + expect(streamable.requests.filter((entry) => entry.message?.method === "tools/call")).toHaveLength(1); + expect(streamable.requests.filter((entry) => + entry.httpMethod === "GET" && entry.headers["last-event-id"] !== undefined + )).toHaveLength(0); + }, 60_000); + test("Streamable HTTP accepts an omitted session and skips DELETE", async () => { streamable = startLegacyStreamableHttpFixture("2025-06-18", { session: false, diff --git a/tests/e2e/mcp-stdio.test.ts b/tests/e2e/mcp-stdio.test.ts index fffdbccee..839dab2be 100644 --- a/tests/e2e/mcp-stdio.test.ts +++ b/tests/e2e/mcp-stdio.test.ts @@ -604,6 +604,40 @@ describe("modern MCP stdio compatibility", () => { return root; }; + // fx reads at most 16 MiB of one stdio message: past that it drops the + // reply and restarts the server. Before, the model was told the server + // went away, as if it had crashed. + test("a stdio reply over fx's message limit tells the model fx restarted the server for it", async () => { + const root = createRoot("oversized-reply", MODERN_FIXTURE); + const profilePath = join(root.home, ".fx", "mcp.json"); + const profile = JSON.parse(readFileSync(profilePath, "utf8")); + profile.mcp.fixture.environment.FX_MCP_RESULT_PAD_BYTES = String(17 * 1024 * 1024); + writeFileSync(profilePath, JSON.stringify(profile)); + let first = ""; + gateway = startFakeGateway([ + fakeGatewayToolCall("select_mcp", "mcp_select_tool", { name: TOOL_NAME }), + fakeGatewayToolCall("call_mcp", TOOL_NAME, { text: "hello" }), + (body: string) => { + first = toolResultText(body, "call_mcp"); + return fakeGatewayFinalText("OVERSIZED_REPLY_DONE"); + }, + ], { + models: [{ id: MODEL, type: "language", tags: ["tool-use"] }], + }); + const result = await runFx(["ask", "--json", "--auto", "--no-save", "Call the MCP tool."], { + cwd: root.workspace, + env: fixtureEnv(root, gateway), + timeoutMs: 40_000, + }); + expect(result.code).toBe(0); + expect(JSON.parse(result.stdout).output).toContain("OVERSIZED_REPLY_DONE"); + expect(first).toContain("its reply was larger than fx reads in one message, so fx restarted the server"); + expect(first).not.toContain("the server went away"); + const wire = readWire(root.wireLogPath); + expect(wire.filter((entry) => entry.message.method === "tools/call")).toHaveLength(1); + await expectFixtureProcessesExited(wire); + }, 60_000); + test("fx ask keeps a big MCP result whole behind a handle, and read_tool_result reaches its end", async () => { const root = bigResultRoot("big-result-stored"); let stored = ""; diff --git a/tests/tla/mcp/HttpLegacy.cfg b/tests/tla/mcp/HttpLegacy.cfg index d722430f5..b016d6063 100644 --- a/tests/tla/mcp/HttpLegacy.cfg +++ b/tests/tla/mcp/HttpLegacy.cfg @@ -17,4 +17,5 @@ INVARIANT ListenRespects405 INVARIANT StreamOnlyWhileOpen PROPERTY EveryRequestEnds PROPERTY ExpiryReinits +PROPERTY TooLargeEnds CHECK_DEADLOCK FALSE diff --git a/tests/tla/mcp/HttpLegacy.mutants.json b/tests/tla/mcp/HttpLegacy.mutants.json index 9842cf787..5397baeb4 100644 --- a/tests/tla/mcp/HttpLegacy.mutants.json +++ b/tests/tla/mcp/HttpLegacy.mutants.json @@ -76,5 +76,11 @@ "old": " /\\ req' = [req EXCEPT ![r] = End(r, \"result\")]\n", "new": " /\\ req' = [req EXCEPT ![r] = [End(r, \"result\") EXCEPT !.stream = TRUE]]\n", "invariant": "StreamOnlyWhileOpen" + }, + { + "description": "An answer over the message limit is resumed like a broken stream", + "old": " /\\ req' = [req EXCEPT ![r] = End(r, \"lost\")]\n /\\ UNCHANGED <>\n\n\\* A status without", + "new": " /\\ req' = [req EXCEPT ![r] = IF req[r].ev /\\ req[r].resumes < MaxResumes\n THEN [@ EXCEPT !.st = \"broken\", !.stream = FALSE]\n ELSE End(r, \"lost\")]\n /\\ UNCHANGED <>\n\n\\* A status without", + "property": "TooLargeEnds" } ] diff --git a/tests/tla/mcp/HttpLegacy.stress.cfg b/tests/tla/mcp/HttpLegacy.stress.cfg index 96df600f1..c8b508122 100644 --- a/tests/tla/mcp/HttpLegacy.stress.cfg +++ b/tests/tla/mcp/HttpLegacy.stress.cfg @@ -17,4 +17,5 @@ INVARIANT ListenRespects405 INVARIANT StreamOnlyWhileOpen PROPERTY EveryRequestEnds PROPERTY ExpiryReinits +PROPERTY TooLargeEnds CHECK_DEADLOCK FALSE diff --git a/tests/tla/mcp/HttpLegacy.tla b/tests/tla/mcp/HttpLegacy.tla index 84ff3e158..c48224b0b 100644 --- a/tests/tla/mcp/HttpLegacy.tla +++ b/tests/tla/mcp/HttpLegacy.tla @@ -14,7 +14,9 @@ \* broke after an event id is resumed with GET and Last-Event-ID once its \* retry wait is over, at most MaxResumes times, and only in the session it \* started in. One without an event id is lost and never re-POSTed -\* (LHTTP-05). A disconnect is not a cancel; cancelling posts +\* (LHTTP-05). An answer over the message limit is lost at once and +\* never resumed: it would come the same again (LHTTP-11). A +\* disconnect is not a cancel; cancelling posts \* notifications/cancelled (LHTTP-08). A request holds a stream only \* while it waits on one: once it ends, answered too, the client closes \* what the server left open (LHTTP-10). @@ -230,6 +232,13 @@ Break(r) == ELSE End(r, "lost")] /\ UNCHANGED <> +\* The answer is over the message limit, on a posted or resumed stream: lost +\* at once, since resuming would only bring it again (LHTTP-11). +TooLarge(r) == + /\ req[r].st \in Open + /\ req' = [req EXCEPT ![r] = End(r, "lost")] + /\ UNCHANGED <> + \* A status without a JSON-RPC answer, other than a session 404: the POST \* failed; a resumed stream is lost (the request may have run). Refused(r) == @@ -278,7 +287,7 @@ Client == Server == \/ InitAnswered(TRUE) \/ InitAnswered(FALSE) \/ InitFailed \/ EndSession - \/ \E r \in Requests : Event(r) \/ Answer(r) \/ Break(r) \/ Refused(r) \/ Expired(r) + \/ \E r \in Requests : Event(r) \/ Answer(r) \/ Break(r) \/ TooLarge(r) \/ Refused(r) \/ Expired(r) \/ ListenRefused \/ ListenBroke \/ ListenExpired Next == Client \/ Server @@ -290,7 +299,7 @@ Fairness == /\ WF_vars(Start) /\ \A r \in Requests : WF_vars(Post(r)) /\ WF_vars(ResumeDue(r)) /\ WF_vars(InitAnswered(TRUE) \/ InitAnswered(FALSE) \/ InitFailed) - /\ \A r \in Requests : WF_vars(Answer(r) \/ Break(r) \/ Refused(r) \/ Expired(r)) + /\ \A r \in Requests : WF_vars(Answer(r) \/ Break(r) \/ TooLarge(r) \/ Refused(r) \/ Expired(r)) Spec == Init /\ [][Next]_vars /\ Fairness @@ -323,6 +332,10 @@ ResumeAfterEventId == \A r \in Requests : req[r].st \in {"broken", "resuming"} => req[r].ev /\ req[r].resumes <= MaxResumes ResumeSameSession == ~crossResume +\* LHTTP-11: an answer over the message limit ends its request, lost; the +\* stream is never resumed for it. An action property, so it adds no state. +TooLargeEnds == [][\A r \in Requests : TooLarge(r) => req'[r].st = "done" /\ req'[r].outcome = "lost"]_vars + \* LHTTP-03: nothing goes out after close, and the DELETE only names a live session. NothingAfterClose == ~sentClosed DeleteOnlyLive == deleted => phase = "closed" /\ sid # 0 /\ sid \notin seenEnd diff --git a/tests/tla/mcp/HttpLegacyTrace.tla b/tests/tla/mcp/HttpLegacyTrace.tla index c5e703351..6e251a62f 100644 --- a/tests/tla/mcp/HttpLegacyTrace.tla +++ b/tests/tla/mcp/HttpLegacyTrace.tla @@ -37,6 +37,7 @@ Step(r, id) == [] r.event = "event" -> Event(id) [] r.event = "answer" -> Answer(id) [] r.event = "break" -> Break(id) + [] r.event = "too_large" -> TooLarge(id) [] r.event = "refused" -> Refused(id) [] r.event = "expired" -> Expired(id) [] r.event = "resume_due" -> ResumeDue(id) diff --git a/tests/tla/mcp/HttpModern.cfg b/tests/tla/mcp/HttpModern.cfg index 659e88a63..7d5f01665 100644 --- a/tests/tla/mcp/HttpModern.cfg +++ b/tests/tla/mcp/HttpModern.cfg @@ -12,3 +12,4 @@ INVARIANT MaybeRanOnlyForLostCalls INVARIANT CancelByClose INVARIANT AnsweredStreamClosed PROPERTY EveryRequestEnds +PROPERTY TooLargeEnds diff --git a/tests/tla/mcp/HttpModern.mutants.json b/tests/tla/mcp/HttpModern.mutants.json index 93fefbc8c..a68a428ac 100644 --- a/tests/tla/mcp/HttpModern.mutants.json +++ b/tests/tla/mcp/HttpModern.mutants.json @@ -58,5 +58,11 @@ "old": " /\\ closed' = TRUE \\* nothing more comes for the answered attempt\n", "new": " /\\ UNCHANGED closed\n", "invariant": "AnsweredStreamClosed" + }, + { + "description": "A listing whose answer was over the message limit is re-issued like a broken stream", + "old": " /\\ phase' = \"lost\"\n /\\ maybeRan' = (kind = \"call\")\n /\\ Deliver(\"lost\")\n /\\ UNCHANGED <>", + "new": " /\\ IF kind = \"idempotent\" /\\ attempts < MaxAttempts\n THEN /\\ attempts' = attempts + 1\n /\\ Post\n /\\ UNCHANGED <>\n ELSE /\\ phase' = \"lost\"\n /\\ maybeRan' = (kind = \"call\")\n /\\ Deliver(\"lost\")\n /\\ UNCHANGED <>\n /\\ UNCHANGED <>", + "property": "TooLargeEnds" } ] diff --git a/tests/tla/mcp/HttpModern.stress.cfg b/tests/tla/mcp/HttpModern.stress.cfg index bd91dd715..917bd906e 100644 --- a/tests/tla/mcp/HttpModern.stress.cfg +++ b/tests/tla/mcp/HttpModern.stress.cfg @@ -12,3 +12,4 @@ INVARIANT MaybeRanOnlyForLostCalls INVARIANT CancelByClose INVARIANT AnsweredStreamClosed PROPERTY EveryRequestEnds +PROPERTY TooLargeEnds diff --git a/tests/tla/mcp/HttpModern.tla b/tests/tla/mcp/HttpModern.tla index f319586d2..4fad3754a 100644 --- a/tests/tla/mcp/HttpModern.tla +++ b/tests/tla/mcp/HttpModern.tla @@ -8,6 +8,8 @@ \* same here), an HTTP failure with no JSON-RPC body, or a stream that \* breaks first. A broken stream is re-issued with a new id, but a tools/call \* is never re-sent by the transport: it ends lost, flagged as maybe run. +\* An answer over the message limit ends the request lost at once, never +\* re-issued, as another attempt would bring the same answer (HTTP-14). \* A -32020 to a tools/call means re-listing the tools and retrying once. \* Cancelling closes the stream; HTTP has no notifications/cancelled (HTTP-08). \* An answer closes its attempt's stream too: the final response ends it, and @@ -110,6 +112,15 @@ StreamBroken == /\ UNCHANGED <> /\ UNCHANGED <> +\* The answer was over the message limit: lost at once, never re-issued, as +\* another attempt would bring the same answer (HTTP-14). A call may have run. +TooLarge == + /\ phase = "posted" + /\ phase' = "lost" + /\ maybeRan' = (kind = "call") + /\ Deliver("lost") + /\ UNCHANGED <> + \* The re-list after -32020 ended: retry the call once, or give up. Relisted(ok) == /\ phase = "relisting" @@ -135,6 +146,7 @@ Environment == \/ \E a \in {"result", "error", "mismatch"} : Answer(a) \/ HttpFailure \/ StreamBroken + \/ TooLarge \/ \E ok \in BOOLEAN : Relisted(ok) Next == Send \/ Environment \/ Cancel @@ -168,6 +180,10 @@ CancelByClose == (phase = "cancelled" => closed) /\ ~cancelNote \* An answered attempt holds no stream, even one the server left open (HTTP-13). AnsweredStreamClosed == phase \in {"done", "relisting"} => closed +\* HTTP-14: an answer over the message limit ends the request, lost, and +\* nothing asks for it again. An action property, so it adds no state. +TooLargeEnds == [][TooLarge => phase' = "lost"]_vars + \* Every request ends. EveryRequestEnds == phase \in {"posted", "relisting"} ~> phase \in Terminal ==== diff --git a/tests/tla/mcp/HttpModernTrace.tla b/tests/tla/mcp/HttpModernTrace.tla index 5e7bd3271..497f0aa16 100644 --- a/tests/tla/mcp/HttpModernTrace.tla +++ b/tests/tla/mcp/HttpModernTrace.tla @@ -22,6 +22,7 @@ Step(r) == [] r.event = "answer" -> Answer(r.data.answer) [] r.event = "http_failure" -> HttpFailure [] r.event = "stream_broken" -> StreamBroken + [] r.event = "too_large" -> TooLarge [] r.event = "relisted" -> Relisted(r.data.ok) [] r.event = "cancel" -> Cancel [] OTHER -> FALSE diff --git a/tests/tla/mcp/rules.md b/tests/tla/mcp/rules.md index 8416e2797..32833d562 100644 --- a/tests/tla/mcp/rules.md +++ b/tests/tla/mcp/rules.md @@ -78,6 +78,7 @@ drivers in `../drivers/mcp/` cite. | HTTP-11 | INFO | Ignore SSE comment lines starting with `:` | `2026-07-28_basic_transports_streamable-http` §Receiving Messages | U06 | | HTTP-12 | INFO | A 403 for a bad `Origin` is not an auth problem. A 404 with `-32601` means the method isn't implemented | `2026-07-28_basic_transports_streamable-http` §Security, §Protocol Version Header | U06, U09 | | HTTP-13 | INFO | Close a request's stream once its final answer is in, even when the server leaves it open: nothing more comes for it, and an open one holds a connection | `2026-07-28_basic_transports_streamable-http` §Receiving Messages ("The final JSON-RPC response SHOULD terminate the stream") | U06 | +| HTTP-14 | INFO | An answer over the client's message limit ends the request lost at once. Re-issuing it (HTTP-09) would bring the same answer, so it isn't; a `tools/call` may have run | `2026-07-28_basic_transports_streamable-http` §Receiving Messages (re-issuing is for a stream that broke before the response) | U06 | ## 2025 HTTP sessions (LHTTP) @@ -93,6 +94,7 @@ drivers in `../drivers/mcp/` cite. | LHTTP-08 | INFO | Over 2025 HTTP, cancel by POSTing `notifications/cancelled`. A disconnect is not a cancellation (LHTTP-05) | `2025-11-25_basic_transports` §Sending Messages; `2025-11-25_basic_utilities_cancellation` §Cancellation Flow | U07 | | LHTTP-09 | MUST | Session ids are visible ASCII (0x21 to 0x7E); clients handle them securely | `2025-11-25_basic_transports` §Session Management | U07 | | LHTTP-10 | INFO | Over 2025 HTTP, close a request's stream once its answer is in, even when the server leaves it open. A disconnect is not a cancellation (LHTTP-05), so this cancels nothing | `2025-11-25_basic_transports` §Sending Messages to the Server ("After the JSON-RPC response has been sent, the server SHOULD terminate the SSE stream") | U07 | +| LHTTP-11 | INFO | An answer over the client's message limit ends the request lost at once. Resuming its stream (LHTTP-05) would bring the same answer, so it isn't resumed | `2025-11-25_basic_transports` §Resumability and Redelivery (resuming is for a stream that disconnected) | U07 | ## Requests, cancellation, and progress (REQ)