Fix/streamable http fixes

src/client/mcp_client.cc

@@ -1010,6 +1010,28 @@ void McpClient::handleClientStreamEvent(ClientStreamEvent event,
 
 void McpClient::resumeAnswer(const std::shared_ptr<RequestContext>& context,
                              const std::string& last_event_id) {
+  if (streamable_session_ && protocol::modern::isModernVersion(
+                                 streamable_session_->protocolVersion())) {
+    // Nothing to pick it up with. This era numbers no events, so there
+    // is no place to carry on from, and it serves no standalone stream
+    // to carry on over — asking for one would be asking for a method
+    // this server refuses, and reading that refusal as a fact about
+    // streams rather than about this request would leave the request
+    // outstanding forever.
+    //
+    // So a stream cut short is the end of what it was carrying. Said
+    // rather than waited on, and said through the ordinary failure path,
+    // which is also what lets go of a subscription.
+    GOPHER_LOG_DEBUG("The stream carrying {} was cut and cannot be resumed",
+                     context->method);
+    stream_recovering_.reset();
+    completeRequestWithError(
+        context, Error(::mcp::jsonrpc::INTERNAL_ERROR,
+                       "The stream carrying this answer was cut off, and this "
+                       "revision has no way to pick one up again"));
+    return;
+  }
+
   if (context->resume_attempts >= config_.streamable_http.resume_attempts) {
     GOPHER_LOG_WARN("Giving up on the answer to {} after {} attempts",
                     context->method, context->resume_attempts);

tests/integration/test_modern_era_subscriptions.cc

@@ -589,5 +589,64 @@ TEST_F(RefusedSubscriptionTest, EndingOneSeversNoOtherRequest) {
          "ending";
 }
 
+// The drop that comes from the far end rather than from here. Its
+// request is still outstanding when the stream goes — which is what the
+// local case never exercises, since ending one takes its request away
+// before the connection closes.
+//
+// A stream cut short in this era is the end of what it was carrying:
+// there are no event ids to carry on from and no standalone stream to
+// carry on over, so asking for one asks for a method this server refuses
+// — and reading that refusal as a fact about streams rather than about
+// this request leaves the request outstanding for good.
+TEST_F(RefusedSubscriptionTest, ADroppedSubscriptionIsEndedNotResumed) {
+  const uint16_t port =
+      server_.start([](const test::Seen& seen) -> test::Reply {
+        if (seen.rpc_method == modern::kMethodServerDiscover) {
+          return test::Reply::write(test::withBody(
+              200, "OK", "application/json",
+              "{\"jsonrpc\":\"2.0\",\"id\":\"d\",\"result\":{\"resultType\":"
+              "\"complete\",\"supportedVersions\":[\"2026-07-28\"],"
+              "\"capabilities\":{}}}",
+              std::string()));
+        }
+        if (seen.rpc_method == modern::kMethodSubscriptionsListen) {
+          return test::Reply::stream(test::streamPrelude());
+        }
+        return test::Reply::write(test::accepted());
+      });
+
+  client::McpClientConfig config;
+  config.client_name = "dropped-subscription-client";
+  config.client_version = "0.0.1";
+  config.num_workers = 1;
+  config.request_timeout = 5000ms;
+  config.protocol_connection_timeout = 5000ms;
+  config.streamable_http.fallback_probe_timeout = 700ms;
+  client_ = client::createMcpClient(config);
+  ASSERT_NE(client_, nullptr);
+  ASSERT_TRUE(holds_alternative<std::nullptr_t>(
+      client_->connect("http://127.0.0.1:" + std::to_string(port) + "/mcp")));
+
+  modern::NotificationFilter what;
+  what.tools_list_changed = true;
+  const int64_t id = client_->listen(what, [](const jsonrpc::Notification&) {});
+  ASSERT_NE(id, 0);
+  ASSERT_TRUE(server_.waitForRpc(modern::kMethodSubscriptionsListen, 1));
+  ASSERT_TRUE(waitUntilTrue(
+      [this]() { return client_->subscriptionsHeld() == 1; }, 3000ms));
+  std::this_thread::sleep_for(300ms);
+
+  const size_t streams_before = server_.countOfMethod("GET");
+  server_.cutStream();
+
+  EXPECT_TRUE(waitUntilTrue(
+      [this]() { return client_->subscriptionsHeld() == 0; }, 3000ms))
+      << "a subscription whose stream was dropped is still being held";
+  EXPECT_EQ(server_.countOfMethod("GET"), streams_before)
+      << "a dropped subscription was asked for again over a stream, in a "
+         "revision that serves none";
+}
+
 }  // namespace
 }  // namespace mcp

tests/interop/child_process.h

@@ -13,12 +13,16 @@
 
 #pragma once
 
+#include <cerrno>
 #include <chrono>
 #include <cstdio>
+#include <cstring>
+#include <poll.h>
 #include <signal.h>
 #include <string>
 #include <thread>
 #include <unistd.h>
+#include <utility>
 #include <vector>
 
 #include <sys/wait.h>
@@ -30,17 +34,43 @@ namespace mcp {
 namespace test {
 
 /** A loopback port the kernel believes is free. */
-inline uint16_t pickFreePort() {
+inline uint16_t pickFreePort(std::string* why_not = nullptr) {
+  auto say = [why_not](const std::string& reason) {
+    if (why_not != nullptr) {
+      *why_not = reason;
+    }
+  };
+
   auto& iface = network::socketInterface();
   auto fd =
       iface.socket(network::SocketType::Stream, network::Address::Type::Ip,
                    network::Address::IpVersion::v4);
   if (!fd.ok()) {
+    say("no socket could be made to find a free port with");
     return 0;
   }
   auto handle = iface.ioHandleForFd(*fd, false);
-  handle->bind(network::Address::parseInternetAddress("127.0.0.1", 0));
-  handle->listen(1);
+
+  // Checked, unlike before. A bind or listen that fails leaves the port
+  // at zero and the caller with nothing to go on — and on a machine
+  // where this is what is wrong, that is the whole of what it needs to
+  // be told.
+  auto bound =
+      handle->bind(network::Address::parseInternetAddress("127.0.0.1", 0));
+  if (!bound.ok()) {
+    say("binding a loopback port failed (errno " + std::to_string(errno) +
+        "); the loopback interface may be down or restricted");
+    handle->close();
+    return 0;
+  }
+  auto listening = handle->listen(1);
+  if (!listening.ok()) {
+    say("listening on a loopback port failed (errno " + std::to_string(errno) +
+        ")");
+    handle->close();
+    return 0;
+  }
+
   auto local = handle->localAddress();
   uint16_t port = 0;
   if (local.ok()) {
@@ -49,10 +79,69 @@ inline uint16_t pickFreePort() {
       port = ip->port();
     }
   }
+  if (port == 0) {
+    say("a loopback port was bound but the kernel did not name it");
+  }
   handle->close();
   return port;
 }
 
+/** What `node --version` says, as major and minor. Zeroes when unknown. */
+inline std::pair<int, int> nodeVersion() {
+  FILE* pipe = popen("node --version 2>/dev/null", "r");
+  if (pipe == nullptr) {
+    return std::make_pair(0, 0);
+  }
+  char buffer[64] = {0};
+  const char* got = fgets(buffer, sizeof(buffer), pipe);
+  pclose(pipe);
+  if (got == nullptr) {
+    return std::make_pair(0, 0);
+  }
+  std::string text(buffer);
+  if (!text.empty() && text[0] == 'v') {
+    text.erase(0, 1);
+  }
+  int major = 0;
+  int minor = 0;
+  if (std::sscanf(text.c_str(), "%d.%d", &major, &minor) != 2) {
+    return std::make_pair(0, 0);
+  }
+  return std::make_pair(major, minor);
+}
+
+/**
+ * What this node needs in order to run a TypeScript file directly.
+ *
+ * Both interop programs are `.ts` run straight by node, which is only
+ * something node can do at all from 22.6, and only without being asked
+ * from 23.6. In between it has to be asked — and a node that is not
+ * asked fails immediately with a syntax error, which from the outside
+ * looks exactly like a server that started and never listened.
+ *
+ * @return False when this node cannot run them however it is asked.
+ */
+inline bool nodeTypeScriptFlags(std::vector<std::string>* flags,
+                                std::string* why_not) {
+  const auto version = nodeVersion();
+  if (version.first == 0) {
+    *why_not = "node is installed but did not say what version it is";
+    return false;
+  }
+  if (version.first < 22 || (version.first == 22 && version.second < 6)) {
+    *why_not = "node " + std::to_string(version.first) + "." +
+               std::to_string(version.second) +
+               " cannot run TypeScript directly; these programs need 22.6 "
+               "or newer";
+    return false;
+  }
+  if (version.first < 23 || (version.first == 23 && version.second < 6)) {
+    // Able, but only when asked.
+    flags->push_back("--experimental-strip-types");
+  }
+  return true;
+}
+
 /** True once something is accepting on the port, or the budget is spent. */
 inline bool waitUntilAccepting(uint16_t port,
                                std::chrono::milliseconds budget) {
@@ -147,47 +236,101 @@ class Child {
     return true;
   }
 
+  /**
+   * How a child ended, for a caller that has to say.
+   *
+   * Three outcomes and not two: a child that was killed is not one that
+   * exited with a code, and neither is one that is still going. Told
+   * apart because the caller that asks is explaining why something did
+   * not start, and "it crashed" is the answer it is most often looking
+   * for.
+   */
+  struct Ending {
+    enum class How { StillRunning, Exited, Signalled };
+
+    How how{How::StillRunning};
+    /** The exit status when it exited, the signal number when it was
+     *  killed, and nothing meaningful while it is still going. */
+    int code{-1};
+
+    bool exitedWith(int expected) const {
+      return how == How::Exited && code == expected;
+    }
+
+    std::string describe() const {
+      if (how == How::Exited) {
+        return "exited with code " + std::to_string(code);
+      }
+      if (how == How::Signalled) {
+        const char* named = strsignal(code);
+        return "was killed by signal " + std::to_string(code) +
+               (named != nullptr ? std::string(" (") + named + ")"
+                                 : std::string());
+      }
+      return "was still running";
+    }
+  };
+
   /**
    * Wait for it to finish, keeping whatever it wrote.
    *
-   * The pipe is drained to EOF first: reaping a child whose output has not
-   * been read is how a test ends up reporting a failure with nothing to
-   * say about it.
+   * Both at once, and neither before the other. Draining to EOF first
+   * looked like the tidy order — read everything, then reap — but EOF on
+   * a child's pipe means every write end has closed, which for a child
+   * that is still running is a thing that has not happened yet. Waiting
+   * for it there is waiting for the child to exit, without a deadline,
+   * inside the call whose whole purpose is to have one. A caller asking
+   * why a peer that is still running has not started would have hung on
+   * the question.
    *
-   * @return Its exit status, or -1 if it did not finish in time.
+   * So the pipe is read for whatever is there, the child is reaped if it
+   * has gone, and the budget is checked — round and round until one of
+   * the last two settles it.
+   *
+   * @return How it ended, and with what. What it wrote up to that point
+   *         is kept whichever way that is — including when it is still
+   *         running, which is what a caller under a deadline gets.
    */
-  int wait(std::chrono::milliseconds budget) {
+  Ending wait(std::chrono::milliseconds budget) {
+    Ending ending;
     if (pid_ <= 0) {
-      return -1;
-    }
-    if (read_fd_ >= 0) {
-      char buffer[4096];
-      ssize_t got = 0;
-      while ((got = read(read_fd_, buffer, sizeof(buffer))) > 0) {
-        output_.append(buffer, static_cast<size_t>(got));
-      }
-      close(read_fd_);
-      read_fd_ = -1;
+      return ending;
     }
 
     const auto deadline = std::chrono::steady_clock::now() + budget;
     int status = 0;
-    while (std::chrono::steady_clock::now() < deadline) {
+    for (;;) {
+      drainWhatIsThere();
+
       const pid_t done = waitpid(pid_, &status, WNOHANG);
       if (done == pid_) {
+        // Gone, so its pipe is at EOF and the rest of what it wrote is
+        // there to be had.
+        drainWhatIsThere();
+        closeRead();
         pid_ = -1;
-        return WIFEXITED(status) ? WEXITSTATUS(status) : -1;
+        if (WIFEXITED(status)) {
+          ending.how = Ending::How::Exited;
+          ending.code = WEXITSTATUS(status);
+        } else if (WIFSIGNALED(status)) {
+          ending.how = Ending::How::Signalled;
+          ending.code = WTERMSIG(status);
+        }
+        return ending;
+      }
+
+      if (std::chrono::steady_clock::now() >= deadline) {
+        return ending;
       }
       std::this_thread::sleep_for(std::chrono::milliseconds(20));
     }
-    return -1;
   }
 
   void stop() {
-    if (read_fd_ >= 0) {
-      close(read_fd_);
-      read_fd_ = -1;
-    }
+    // Whatever it managed to say before being told to go, since a child
+    // that has to be killed is one somebody will want an account of.
+    drainWhatIsThere();
+    closeRead();
     if (pid_ <= 0) {
       return;
     }
@@ -207,6 +350,43 @@ class Child {
 
   bool running() const { return pid_ > 0; }
 
+ private:
+  /** Take what has been written so far, and never wait for more. */
+  void drainWhatIsThere() {
+    if (read_fd_ < 0) {
+      return;
+    }
+    for (;;) {
+      struct pollfd waiting;
+      waiting.fd = read_fd_;
+      waiting.events = POLLIN;
+      waiting.revents = 0;
+      // No timeout at all: this asks what is there, never what is
+      // coming.
+      const int ready = poll(&waiting, 1, 0);
+      if (ready <= 0) {
+        return;
+      }
+      char buffer[4096];
+      const ssize_t got = read(read_fd_, buffer, sizeof(buffer));
+      if (got > 0) {
+        output_.append(buffer, static_cast<size_t>(got));
+        continue;
+      }
+      // Zero is EOF and negative is nothing more to be had now; either
+      // way there is nothing further to read in this pass.
+      return;
+    }
+  }
+
+  void closeRead() {
+    if (read_fd_ >= 0) {
+      close(read_fd_);
+      read_fd_ = -1;
+    }
+  }
+
+ public:
   /** Everything it wrote, once wait() has read it. */
   const std::string& output() const { return output_; }