diff --git a/proto/ETerminal.proto b/proto/ETerminal.proto
index 6b831869c..deaa5b3fa 100644
--- a/proto/ETerminal.proto
+++ b/proto/ETerminal.proto
@@ -57,10 +57,25 @@ message PortForwardData {
   optional bool closed = 5;
 }
 
+enum FlowControlMode {
+  // Legacy behavior: no application-level buffering or kernel buffer
+  // tuning. This is the default and what servers assume for old clients
+  // that don't send flow_control_mode: current users are unchanged.
+  FLOW_CONTROL_NONE = 0;
+  // Bounded buffering; when full, stop reading from the source. Lossless
+  // (like plain ssh, but with small, tuned buffers so Ctrl-C stays
+  // responsive on a saturated link).
+  FLOW_CONTROL_BACKPRESSURE = 1;
+  // Bounded buffering; when full, drop the oldest pending output. The
+  // producing process never stalls and the display stays near real time.
+  FLOW_CONTROL_DISCARD = 2;
+}
+
 message InitialPayload {
   optional bool jumphost = 1 [default = false];
   repeated PortForwardSourceRequest reversetunnels = 2;
   map<string, string> environmentvariables = 3;
+  optional FlowControlMode flow_control_mode = 4 [default = FLOW_CONTROL_NONE];
 }
 
 message InitialResponse {
@@ -75,6 +90,9 @@ message ConfigParams {
 message TermInit {
   repeated string environmentnames = 1;
   repeated string environmentvalues = 2;
+  // Tells etterminal whether the attached client opted into flow control,
+  // so it can shrink the kernel buffer on its socket to the server.
+  optional FlowControlMode flow_control_mode = 3 [default = FLOW_CONTROL_NONE];
 }
 
 message TerminalUserInfo {
diff --git a/src/base/Headers.hpp b/src/base/Headers.hpp
index c3c5468f8..d85f8eb3b 100644
--- a/src/base/Headers.hpp
+++ b/src/base/Headers.hpp
@@ -367,6 +367,12 @@ inline bool waitOnSocketData(int fd) {
     if (errno == EINTR) {
       // Interrupted by the signal, the caller will retry.
       return false;
+    } else if (errno == EBADF || errno == EINVAL) {
+      // The fd was closed under us (e.g. peer disconnected). It will never
+      // become readable, and callers use this helper in retry loops, so
+      // returning false would spin at 100% CPU. Throw so the loop treats
+      // it as socket death.
+      throw std::runtime_error("Socket closed while waiting for data");
     } else {
       FATAL_FAIL(selectResult);
     }
@@ -389,8 +395,12 @@ inline bool isSocketWritable(int fd) {
   tv.tv_usec = 0;
   const int selectResult = select(fd + 1, NULL, &fdset, NULL, &tv);
   if (selectResult < 0) {
-    if (errno == EINTR) {
-      // Interrupted by the signal, the caller will retry.
+    if (errno == EINTR || errno == EBADF || errno == EINVAL) {
+      // EINTR: interrupted by a signal. EBADF/EINVAL: the fd was closed
+      // under us (e.g. client disconnected). Returning false is safe in
+      // both cases: callers use this as a non-blocking "can I write more?"
+      // poll and stop on false; a dead socket is then surfaced by the next
+      // read/write on the connection.
       return false;
     } else {
       FATAL_FAIL(selectResult);
diff --git a/src/base/PipeSocketHandler.cpp b/src/base/PipeSocketHandler.cpp
index 509a05d65..0702f6188 100644
--- a/src/base/PipeSocketHandler.cpp
+++ b/src/base/PipeSocketHandler.cpp
@@ -144,4 +144,18 @@ void PipeSocketHandler::stopListening(const SocketEndpoint& endpoint) {
   FATAL_FAIL(::close(sockFd));
 #endif
 }
+
+void PipeSocketHandler::minimizeKernelBuffering(int fd) {
+#ifndef WIN32
+  // Bound the kernel buffer on this unix socket. Terminal output waits
+  // here when the server applies backpressure (stops reading); the default
+  // (~200KB on Linux) is seconds of stale output on a slow link. 64KB does
+  // not limit throughput on a local socket.
+  int sndbuf = 64 * 1024;
+  if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, (char*)&sndbuf, sizeof(sndbuf)) <
+      0) {
+    LOG(WARNING) << "Failed to set SO_SNDBUF: " << strerror(errno);
+  }
+#endif
+}
 }  // namespace et
diff --git a/src/base/PipeSocketHandler.hpp b/src/base/PipeSocketHandler.hpp
index 2afb99e55..7a2199574 100644
--- a/src/base/PipeSocketHandler.hpp
+++ b/src/base/PipeSocketHandler.hpp
@@ -30,6 +30,8 @@ class PipeSocketHandler : public UnixSocketHandler {
    */
   virtual void stopListening(const SocketEndpoint& endpoint);
 
+  virtual void minimizeKernelBuffering(int fd);
+
  protected:
   /** @brief Tracks path -> listening socket descriptors for each pipe. */
   map<string, set<int>> pipeServerSockets;
diff --git a/src/base/SocketHandler.hpp b/src/base/SocketHandler.hpp
index 232c32969..168fb16ca 100644
--- a/src/base/SocketHandler.hpp
+++ b/src/base/SocketHandler.hpp
@@ -20,6 +20,15 @@ class SocketHandler {
    * descriptor.
    */
   virtual bool hasData(int fd) = 0;
+
+  /**
+   * @brief Shrinks the kernel's outgoing queue on fd so that pending data
+   * waits in application-level buffers, where flow control can apply
+   * backpressure or discard stale output. Only called for sessions that
+   * opt into flow control (--flow-control backpressure|discard); the
+   * default is a no-op so legacy sessions keep stock kernel buffering.
+   */
+  virtual void minimizeKernelBuffering(int fd) {}
   /**
    * @brief Reads up to count bytes from fd.
    */
diff --git a/src/base/TcpSocketHandler.cpp b/src/base/TcpSocketHandler.cpp
index da8a90ebe..a8feeb6ce 100644
--- a/src/base/TcpSocketHandler.cpp
+++ b/src/base/TcpSocketHandler.cpp
@@ -293,4 +293,23 @@ void TcpSocketHandler::initSocket(int fd) {
         fd, SOL_SOCKET, SO_LINGER, (const char*)&so_linger, sizeof so_linger));
   }
 }
+
+void TcpSocketHandler::minimizeKernelBuffering(int fd) {
+#ifdef TCP_NOTSENT_LOWAT
+  // Keep the kernel's not-yet-sent queue small so pending terminal output
+  // stays in application-level buffers, where flow control can apply
+  // backpressure or discard stale data. Without this, select()/poll()
+  // report the socket writable whenever the autotuned (multi-MB) send
+  // buffer has room, and on a slow link many seconds of stale output pile
+  // up in the kernel where they can no longer be dropped (issue #631).
+  // TCP_NOTSENT_LOWAT only limits *unsent* data; sent-but-unacked
+  // (in-flight) data is unaffected, so throughput on high-BDP links is
+  // preserved. Best effort: not supported on all platforms.
+  int lowat = 32 * 1024;
+  if (setsockopt(fd, IPPROTO_TCP, TCP_NOTSENT_LOWAT, (char*)&lowat,
+                 sizeof(lowat)) < 0) {
+    LOG(WARNING) << "Failed to set TCP_NOTSENT_LOWAT: " << strerror(errno);
+  }
+#endif
+}
 }  // namespace et
diff --git a/src/base/TcpSocketHandler.hpp b/src/base/TcpSocketHandler.hpp
index d81decf30..9f657eee9 100644
--- a/src/base/TcpSocketHandler.hpp
+++ b/src/base/TcpSocketHandler.hpp
@@ -31,6 +31,8 @@ class TcpSocketHandler : public UnixSocketHandler {
    */
   virtual void stopListening(const SocketEndpoint& endpoint);
 
+  virtual void minimizeKernelBuffering(int fd);
+
  protected:
   /** @brief Tracks all listening sockets created per TCP port. */
   map<int, set<int>> portServerSockets;
diff --git a/src/base/WriteBuffer.hpp b/src/base/WriteBuffer.hpp
new file mode 100644
index 000000000..1dd39c937
--- /dev/null
+++ b/src/base/WriteBuffer.hpp
@@ -0,0 +1,150 @@
+#ifndef __ET_WRITE_BUFFER__
+#define __ET_WRITE_BUFFER__
+
+#include "Headers.hpp"
+
+namespace et {
+
+enum class WriteBufferMode {
+  BACKPRESSURE,  // Stop accepting data when buffer is full
+  DISCARD        // Drop oldest data when buffer is full
+};
+
+/**
+ * @brief Bounded buffer for pending write data, enabling flow control.
+ *
+ * Supports two modes:
+ * - BACKPRESSURE: canAcceptMore() returns false when full, causing callers
+ *   to stop reading from the source. Processes stall when consumer is slow.
+ * - DISCARD: canAcceptMore() always returns true. When the buffer exceeds
+ *   MAX_BUFFER_SIZE, oldest data is dropped. Processes never stall.
+ */
+class WriteBuffer {
+ public:
+  /** @brief Maximum bytes to buffer before applying backpressure or
+   * discarding. Every byte held here (or in any buffer downstream) is stale
+   * data the user must wait through on a slow link, so keep this small: at
+   * 100KB/s, 64KB is ~0.6s of output. It only needs to absorb short bursts
+   * between drain opportunities (reads are 16KB), not smooth over the
+   * network. This is a soft bound: canAcceptMore() admits a whole chunk
+   * whenever size() is below the limit, so the buffer can exceed it by up
+   * to one chunk. */
+  static constexpr size_t MAX_BUFFER_SIZE = 64 * 1024;  // 64KB
+
+  explicit WriteBuffer(WriteBufferMode mode = WriteBufferMode::BACKPRESSURE)
+      : mode(mode), totalBytes(0), writeOffset(0) {}
+
+  /**
+   * @brief Returns true if the buffer has room for more data.
+   * In BACKPRESSURE mode, returns false when buffer >= MAX_BUFFER_SIZE.
+   * In DISCARD mode, always returns true (old data will be dropped).
+   */
+  bool canAcceptMore() const {
+    if (mode == WriteBufferMode::DISCARD) return true;
+    return totalBytes < MAX_BUFFER_SIZE;
+  }
+
+  /**
+   * @brief Returns true if there is data waiting to be written.
+   */
+  bool hasPendingData() const { return !pending.empty(); }
+
+  /**
+   * @brief Returns the current amount of buffered data in bytes.
+   */
+  size_t size() const { return totalBytes; }
+
+  /**
+   * @brief Returns the buffer mode.
+   */
+  WriteBufferMode getMode() const { return mode; }
+
+  /**
+   * @brief Adds data to the end of the buffer.
+   * In DISCARD mode, drops oldest chunks if the buffer would exceed
+   * MAX_BUFFER_SIZE.
+   * @param data The data to enqueue.
+   */
+  void enqueue(const string& data) {
+    if (data.empty()) return;
+    pending.push_back(data);
+    totalBytes += data.size();
+
+    if (mode == WriteBufferMode::DISCARD) {
+      discardOldest();
+    }
+  }
+
+  /**
+   * @brief Returns a pointer to the next bytes to write and the count.
+   * @param count Output: number of bytes available for writing.
+   * @return Pointer to the data, or nullptr if buffer is empty.
+   */
+  const char* peekData(size_t* count) const {
+    if (pending.empty()) {
+      *count = 0;
+      return nullptr;
+    }
+    const string& front = pending.front();
+    *count = front.size() - writeOffset;
+    return front.data() + writeOffset;
+  }
+
+  /**
+   * @brief Removes bytesWritten from the front of the buffer.
+   * @param bytesWritten Number of bytes successfully written to socket.
+   */
+  void consume(size_t bytesWritten) {
+    if (bytesWritten == 0) return;
+
+    while (bytesWritten > 0 && !pending.empty()) {
+      string& front = pending.front();
+      size_t available = front.size() - writeOffset;
+
+      if (bytesWritten >= available) {
+        // Consumed the entire front chunk
+        bytesWritten -= available;
+        totalBytes -= available;
+        writeOffset = 0;
+        pending.pop_front();
+      } else {
+        // Partial consumption
+        writeOffset += bytesWritten;
+        totalBytes -= bytesWritten;
+        bytesWritten = 0;
+      }
+    }
+  }
+
+  /**
+   * @brief Clears all pending data.
+   */
+  void clear() {
+    pending.clear();
+    totalBytes = 0;
+    writeOffset = 0;
+  }
+
+ private:
+  /**
+   * @brief Drops oldest chunks until buffer is within MAX_BUFFER_SIZE.
+   * Only called in DISCARD mode after enqueue.
+   */
+  void discardOldest() {
+    while (totalBytes > MAX_BUFFER_SIZE && !pending.empty()) {
+      string& front = pending.front();
+      size_t frontSize = front.size() - writeOffset;
+      totalBytes -= frontSize;
+      writeOffset = 0;
+      pending.pop_front();
+    }
+  }
+
+  WriteBufferMode mode;
+  std::deque<string> pending;
+  size_t totalBytes;
+  size_t writeOffset;  // Offset into the front chunk for partial writes
+};
+}  // namespace et
+
+#endif  // __ET_WRITE_BUFFER__
diff --git a/src/terminal/TerminalClient.cpp b/src/terminal/TerminalClient.cpp
index b3c777fb4..2df893b10 100644
--- a/src/terminal/TerminalClient.cpp
+++ b/src/terminal/TerminalClient.cpp
@@ -2,6 +2,7 @@
 
 #include "TelemetryService.hpp"
 #include "TunnelUtils.hpp"
+#include "WriteBuffer.hpp"
 
 namespace et {
 
@@ -12,14 +13,20 @@ TerminalClient::TerminalClient(
     const string& passkey, shared_ptr<Console> _console, bool jumphost,
     const string& tunnels, const string& reverseTunnels, bool forwardSshAgent,
     const string& identityAgent, int _keepaliveDuration,
-    const vector<pair<string, string>>& envVars)
+    const vector<pair<string, string>>& envVars,
+    et::FlowControlMode _flowControlMode)
     : console(_console),
       shuttingDown(false),
-      keepaliveDuration(_keepaliveDuration) {
+      keepaliveDuration(_keepaliveDuration),
+      flowControlMode(_flowControlMode) {
   portForwardHandler = shared_ptr<PortForwardHandler>(
       new PortForwardHandler(_socketHandler, _pipeSocketHandler));
   InitialPayload payload;
   payload.set_jumphost(jumphost);
+  if (flowControlMode != et::FLOW_CONTROL_NONE) {
+    // Left unset for NONE so the wire format matches old clients exactly.
+    payload.set_flow_control_mode(flowControlMode);
+  }
 
   for (const auto& envVar : envVars) {
     (*payload.mutable_environmentvariables())[envVar.first] = envVar.second;
@@ -172,6 +179,14 @@ void TerminalClient::run(const string& command, const bool noexit) {
 
   TerminalInfo lastTerminalInfo;
 
+  const bool flowControlEnabled = (flowControlMode != et::FLOW_CONTROL_NONE);
+  // Flow control (opt-in): terminal output is staged here and written to the
+  // console at the top of each loop iteration. Unused when flow control is
+  // off: output is written straight to the console (legacy behavior).
+  WriteBuffer consoleOutputBuffer(flowControlMode == et::FLOW_CONTROL_DISCARD
+                                      ? WriteBufferMode::DISCARD
+                                      : WriteBufferMode::BACKPRESSURE);
+
   if (!console.get()) {
     // NOTE: ../../scripts/ssh-et relies on the wording of this message, so if
     // you change it please update it as well.
@@ -200,8 +215,12 @@ void TerminalClient::run(const string& command, const bool noexit) {
     }
     int clientFd = connection->getSocketFd();
     if (clientFd > 0) {
-      FD_SET(clientFd, &rfd);
-      maxfd = max(maxfd, clientFd);
+      // Only read from the server if the console output buffer has room;
+      // this propagates backpressure when the console is slow.
+      if (!flowControlEnabled || consoleOutputBuffer.canAcceptMore()) {
+        FD_SET(clientFd, &rfd);
+        maxfd = max(maxfd, clientFd);
+      }
     }
     // Include port forward sockets in select for low-latency forwarding.
     set<int> pfFds;
@@ -215,6 +234,24 @@ void TerminalClient::run(const string& command, const bool noexit) {
     select(maxfd + 1, &rfd, NULL, NULL, &tv);
 
     try {
+      // First, drain buffered terminal output to the console. Coalesce all
+      // pending chunks into a single write to avoid intermediate renders
+      // (flicker), matching the non-buffered path below.
+      if (console && consoleOutputBuffer.hasPendingData()) {
+        string pending;
+        size_t count;
+        const char* data;
+        while ((data = consoleOutputBuffer.peekData(&count)) != nullptr &&
+               count > 0) {
+          pending.append(data, count);
+          consoleOutputBuffer.consume(count);
+        }
+        if (!pending.empty()) {
+          // May block if the console is slow; bounded by the buffer size.
+          console->write(pending);
+        }
+      }
+
       if (console) {
         // Check for data to send.
         if (FD_ISSET(consoleFd, &rfd)) {
@@ -287,7 +324,12 @@ void TerminalClient::run(const string& command, const bool noexit) {
         // arriving in one write followed by the repaint in the next), which
         // produces visible flicker.
         string coalesced;
-        while (connection->hasData()) {
+        // Stop pulling packets once the console buffer is full: in
+        // backpressure mode it would otherwise grow past its bound by
+        // however much the kernel had queued. Unprocessed packets stay in
+        // the socket buffer until the next iteration.
+        while (connection->hasData() &&
+               (!flowControlEnabled || consoleOutputBuffer.canAcceptMore())) {
           VLOG(4) << "connection has data";
           Packet packet;
           if (!connection->read(&packet)) {
@@ -326,17 +368,26 @@ void TerminalClient::run(const string& command, const bool noexit) {
           }
         }
         if (console && !coalesced.empty()) {
-          console->write(coalesced);
+          if (flowControlEnabled) {
+            // Staged for the flow-controlled drain at the top of the loop
+            consoleOutputBuffer.enqueue(coalesced);
+          } else {
+            console->write(coalesced);
+          }
         }
       }
 
       if (clientFd > 0 && keepaliveTime < time(NULL)) {
         keepaliveTime = time(NULL) + keepaliveDuration;
-        if (waitingOnKeepalive) {
+        // While the console buffer is full we are intentionally not reading
+        // the connection, so the keepalive echo may be sitting unread in the
+        // socket; don't treat that as a dead connection.
+        if (waitingOnKeepalive &&
+            !(flowControlEnabled && consoleOutputBuffer.hasPendingData())) {
           LOG(INFO) << "Missed a keepalive, killing connection.";
           connection->closeSocketAndMaybeReconnect();
           waitingOnKeepalive = false;
-        } else {
+        } else if (!waitingOnKeepalive) {
           LOG(INFO) << "Writing keepalive packet";
           connection->writePacket(Packet(TerminalPacketType::KEEP_ALIVE, ""));
           waitingOnKeepalive = true;
diff --git a/src/terminal/TerminalClient.hpp b/src/terminal/TerminalClient.hpp
index a56b635f8..9906af129 100644
--- a/src/terminal/TerminalClient.hpp
+++ b/src/terminal/TerminalClient.hpp
@@ -32,7 +32,8 @@ class TerminalClient {
                  bool jumphost, const string& tunnels,
                  const string& reverseTunnels, bool forwardSshAgent,
                  const string& identityAgent, int _keepaliveDuration,
-                 const vector<pair<string, string>>& envVars);
+                 const vector<pair<string, string>>& envVars,
+                 et::FlowControlMode _flowControlMode = et::FLOW_CONTROL_NONE);
   /** @brief Tears down the client, closing sockets and stopping background
    * threads. */
   virtual ~TerminalClient();
@@ -60,6 +61,8 @@ class TerminalClient {
   recursive_mutex shutdownMutex;
   /** @brief Keepalive interval (seconds) sent to the server. */
   int keepaliveDuration;
+  /** @brief Flow control mode for output buffering. */
+  et::FlowControlMode flowControlMode;
 };
 
 }  // namespace et
diff --git a/src/terminal/TerminalClientMain.cpp b/src/terminal/TerminalClientMain.cpp
index 570bfe806..ced425ba0 100644
--- a/src/terminal/TerminalClientMain.cpp
+++ b/src/terminal/TerminalClientMain.cpp
@@ -185,7 +185,20 @@ int main(int argc, char** argv) {
          "If set, communicate to etserver on the matching fifo name",
          cxxopts::value<std::string>()->default_value(""))  //
         ("ssh-option", "Options to pass down to `ssh -o`",
-         cxxopts::value<std::vector<std::string>>());
+         cxxopts::value<std::vector<std::string>>())  //
+        ("idpasskey",
+         "If set, skip SSH and use this id/passkey directly (format: "
+         "id/passkey). Start etterminal manually before connecting.",
+         cxxopts::value<std::string>())  //
+        ("flow-control",
+         "Flow control mode: 'none' (default; legacy behavior, unchanged), "
+         "'backpressure' (lossless: bounded, tuned buffers keep Ctrl-C "
+         "responsive on a slow link; when the client can't keep up the "
+         "remote process is paused, like plain ssh), or 'discard' (drop "
+         "the oldest pending output so the remote process never stalls "
+         "and the display stays close to real time; not for consumers "
+         "that require every byte, e.g. tmux -CC)",
+         cxxopts::value<std::string>()->default_value("none"));
 
     options.parse_positional({"host"});
     auto result = options.parse(argc, argv);
@@ -291,7 +304,7 @@ int main(int argc, char** argv) {
       exit(0);
     }
 
-    {
+    if (!result.count("idpasskey")) {
       char* home_dir = ssh_get_user_home_dir();
       const char* host_from_command = destinationHost.c_str();
       ssh_options_set(&sshConfigOptions, SSH_OPTIONS_HOST,
@@ -369,7 +382,7 @@ int main(int argc, char** argv) {
     shared_ptr<SocketHandler> clientSocket(new TcpSocketHandler());
     shared_ptr<SocketHandler> clientPipeSocket(new PipeSocketHandler());
 
-    if (!ping(socketEndpoint, clientSocket)) {
+    if (!result.count("idpasskey") && !ping(socketEndpoint, clientSocket)) {
       CLOG(INFO, "stdout") << "Could not reach the ET server: "
                            << socketEndpoint.name() << ":"
                            << socketEndpoint.port() << endl;
@@ -432,17 +445,42 @@ int main(int argc, char** argv) {
       }
     }
 
-    auto subprocessUtils = make_shared<SubprocessUtils>();
-    SshSetupHandler sshSetupHandler(subprocessUtils);
-    pair<string, string> idpasskeypair = sshSetupHandler.SetupSsh(
-        username, destinationHost, host_alias, destinationPort, jumphost,
-        jServerFifo, result.count("x") > 0, result["verbose"].as<int>(),
-        etterminal_path, serverFifo, ssh_options);
+    pair<string, string> idpasskeypair;
+    if (result.count("idpasskey")) {
+      auto tokens = split(result["idpasskey"].as<string>(), '/');
+      if (tokens.size() != 2 || tokens[0].empty() || tokens[1].empty()) {
+        CLOG(INFO, "stdout")
+            << "Invalid --idpasskey format. Expected: id/passkey" << endl;
+        exit(1);
+      }
+      idpasskeypair = {tokens[0], tokens[1]};
+    } else {
+      auto subprocessUtils = make_shared<SubprocessUtils>();
+      SshSetupHandler sshSetupHandler(subprocessUtils);
+      idpasskeypair = sshSetupHandler.SetupSsh(
+          username, destinationHost, host_alias, destinationPort, jumphost,
+          jServerFifo, result.count("x") > 0, result["verbose"].as<int>(),
+          etterminal_path, serverFifo, ssh_options);
+    }
+
+    et::FlowControlMode flowControlMode = et::FLOW_CONTROL_NONE;
+    string flowControlStr = result["flow-control"].as<string>();
+    if (flowControlStr == "backpressure") {
+      flowControlMode = et::FLOW_CONTROL_BACKPRESSURE;
+    } else if (flowControlStr == "discard") {
+      flowControlMode = et::FLOW_CONTROL_DISCARD;
+    } else if (flowControlStr != "none") {
+      CLOG(INFO, "stdout") << "Invalid flow-control mode: " << flowControlStr
+                           << ". Must be 'none', 'backpressure', or 'discard'."
+                           << endl;
+      exit(1);
+    }
 
     TerminalClient terminalClient(
         clientSocket, clientPipeSocket, socketEndpoint, idpasskeypair.first,
         idpasskeypair.second, console, is_jumphost, tunnel_arg, r_tunnel_arg,
-        forwardAgent, sshSocket, keepaliveDuration, sshConfigOptions.env_vars);
+        forwardAgent, sshSocket, keepaliveDuration, sshConfigOptions.env_vars,
+        flowControlMode);
     terminalClient.run(
         result.count("command") ? result["command"].as<string>() : "",
         result.count("noexit"));
diff --git a/src/terminal/TerminalMain.cpp b/src/terminal/TerminalMain.cpp
index 173f27cfa..1feed6d31 100644
--- a/src/terminal/TerminalMain.cpp
+++ b/src/terminal/TerminalMain.cpp
@@ -124,7 +124,7 @@ int main(int argc, char** argv) {
         STFATAL << "Invalid number of tokens: " << tokens.size();
       }
     } else {
-      string idpasskey = result["idpasskey"].as<string>();
+      idpasskey = result["idpasskey"].as<string>();
       if (result.count("idpasskeyfile")) {
         // Check for passkey file
         std::ifstream t(result["idpasskeyfile"].as<string>().c_str());
diff --git a/src/terminal/TerminalServer.cpp b/src/terminal/TerminalServer.cpp
index b86ac1e09..9ba035315 100644
--- a/src/terminal/TerminalServer.cpp
+++ b/src/terminal/TerminalServer.cpp
@@ -2,6 +2,7 @@
 #include "TerminalServer.hpp"
 
 #include "TelemetryService.hpp"
+#include "WriteBuffer.hpp"
 
 #define BUF_SIZE (16 * 1024)
 
@@ -124,6 +125,21 @@ void TerminalServer::runJumpHost(
       terminalFd,
       Packet(TerminalPacketType::JUMPHOST_INIT, protoToString(payload)));
 
+  // Flow control (only when the client opted in): queue of pending jumphost
+  // packets for the client.
+  const et::FlowControlMode jumphostMode = payload.flow_control_mode();
+  const bool jumphostFlowControlEnabled =
+      (jumphostMode != et::FLOW_CONTROL_NONE);
+  const bool jumphostDiscard = (jumphostMode == et::FLOW_CONTROL_DISCARD);
+  std::deque<Packet> pendingPackets;
+  size_t pendingBytes = 0;
+  // Bytes of TERMINAL_BUFFER packets in pendingPackets. Only those are
+  // droppable in discard mode; when the queue is dominated by non-droppable
+  // control packets (e.g. port-forward data), reads must stop so the queue
+  // stays bounded.
+  size_t droppableBytes = 0;
+  const size_t MAX_PENDING_BYTES = WriteBuffer::MAX_BUFFER_SIZE;
+
   while (true) {
     {
       lock_guard<std::mutex> guard(terminalThreadMutex);
@@ -132,33 +148,103 @@ void TerminalServer::runJumpHost(
       }
     }
 
-    fd_set rfd;
+    fd_set rfd, wfd;
     timeval tv;
 
     FD_ZERO(&rfd);
+    FD_ZERO(&wfd);
     int maxfd = -1;
     // Only drain the terminal while the client connection can absorb the
     // data, so backpressure reaches the terminal instead of this loop
-    // blocking inside writePacket()
-    if (serverClientState->canBufferWrite(2 * BUF_SIZE)) {
+    // blocking inside writePacket(). In flow-control modes the bounded
+    // pending queue gates reads instead; in discard mode room can always be
+    // made by dropping old terminal output, so only the non-droppable
+    // (control) backlog gates reading.
+    bool readTerminal;
+    if (!jumphostFlowControlEnabled) {
+      readTerminal = serverClientState->canBufferWrite(2 * BUF_SIZE);
+    } else if (jumphostDiscard) {
+      readTerminal = (pendingBytes - droppableBytes) < MAX_PENDING_BYTES;
+    } else {
+      readTerminal = pendingBytes < MAX_PENDING_BYTES;
+    }
+    if (readTerminal) {
       FD_SET(terminalFd, &rfd);
       maxfd = terminalFd;
     }
     int serverClientFd = serverClientState->getSocketFd();
     if (serverClientFd > 0) {
+      if (jumphostFlowControlEnabled) {
+        // Reapply every iteration; see runTerminal for why fd tracking is
+        // not reliable across reconnects.
+        getSocketHandler()->minimizeKernelBuffering(serverClientFd);
+      }
       FD_SET(serverClientFd, &rfd);
       maxfd = max(maxfd, serverClientFd);
+
+      // Wake as soon as the socket can take more pending packets.
+      if (!pendingPackets.empty()) {
+        FD_SET(serverClientFd, &wfd);
+      }
     }
     tv.tv_sec = 0;
     tv.tv_usec = 100000;
-    select(maxfd + 1, &rfd, NULL, NULL, &tv);
+    if (select(maxfd + 1, &rfd, &wfd, NULL, &tv) < 0 && errno == EINTR) {
+      // See runTerminal: retry only on EINTR; other errors fall through so
+      // the read/write paths surface the dead fd.
+      continue;
+    }
 
     try {
+      // Drain pending packets while the socket can take more
+      if (serverClientFd > 0 && FD_ISSET(serverClientFd, &wfd) &&
+          !pendingPackets.empty()) {
+        while (!pendingPackets.empty()) {
+          serverClientState->writePacket(pendingPackets.front());
+          pendingBytes -= pendingPackets.front().length();
+          if (pendingPackets.front().getHeader() ==
+              TerminalPacketType::TERMINAL_BUFFER) {
+            droppableBytes -= pendingPackets.front().length();
+          }
+          pendingPackets.pop_front();
+
+          if (!isSocketWritable(serverClientFd)) {
+            break;  // Kernel queue is full enough; stop draining
+          }
+        }
+      }
+
       if (FD_ISSET(terminalFd, &rfd)) {
         try {
           Packet packet;
           if (terminalSocketHandler->readPacket(terminalFd, &packet)) {
-            serverClientState->writePacket(packet);
+            if (!jumphostFlowControlEnabled) {
+              serverClientState->writePacket(packet);
+            } else {
+              pendingPackets.push_back(packet);
+              pendingBytes += packet.length();
+              if (packet.getHeader() == TerminalPacketType::TERMINAL_BUFFER) {
+                droppableBytes += packet.length();
+              }
+
+              // In discard mode, drop the oldest droppable packets when
+              // over the limit. Only terminal output is safe to drop:
+              // control packets (port forwarding, responses, keepalives)
+              // must be delivered or the protocol state desyncs.
+              if (jumphostDiscard) {
+                auto it = pendingPackets.begin();
+                while (pendingBytes > MAX_PENDING_BYTES &&
+                       it != pendingPackets.end()) {
+                  if (it->getHeader() == TerminalPacketType::TERMINAL_BUFFER) {
+                    pendingBytes -= it->length();
+                    droppableBytes -= it->length();
+                    it = pendingPackets.erase(it);
+                  } else {
+                    ++it;
+                  }
+                }
+              }
+            }
           }
         } catch (const std::runtime_error& ex) {
           LOG(INFO) << "Terminal session ended" << ex.what();
@@ -263,7 +349,13 @@ void TerminalServer::runTerminal(
   shared_ptr<SocketHandler> terminalSocketHandler =
       terminalRouter->getSocketHandler();
 
+  const et::FlowControlMode flowControlMode = payload.flow_control_mode();
+  const bool flowControlEnabled = (flowControlMode != et::FLOW_CONTROL_NONE);
+
   TermInit termInit;
+  if (flowControlEnabled) {
+    termInit.set_flow_control_mode(flowControlMode);
+  }
   for (auto& it : environmentVariables) {
     *(termInit.add_environmentnames()) = it.first;
     *(termInit.add_environmentvalues()) = it.second;
@@ -272,6 +364,14 @@ void TerminalServer::runTerminal(
       terminalFd,
       Packet(TerminalPacketType::TERMINAL_INIT, protoToString(termInit)));
 
+  // Flow control (opt-in): terminal output is staged here and drained only
+  // while the socket reports writable, so the backlog stays where it can be
+  // gated (backpressure) or dropped (discard) instead of piling up in the
+  // kernel. Unused when the client didn't opt in (FLOW_CONTROL_NONE).
+  WriteBuffer terminalOutputBuffer(flowControlMode == et::FLOW_CONTROL_DISCARD
+                                       ? WriteBufferMode::DISCARD
+                                       : WriteBufferMode::BACKPRESSURE);
+
   while (run) {
     {
       lock_guard<std::mutex> guard(terminalThreadMutex);
@@ -282,22 +382,38 @@ void TerminalServer::runTerminal(
 
     // Data structures needed for select() and
     // non-blocking I/O.
-    fd_set rfd;
+    fd_set rfd, wfd;
     timeval tv;
 
     FD_ZERO(&rfd);
+    FD_ZERO(&wfd);
     int maxfd = -1;
     // Only drain the terminal while the client connection can absorb the
     // data, so backpressure reaches the shell instead of this loop blocking
-    // inside writePacket()
-    if (serverClientState->canBufferWrite(2 * BUF_SIZE)) {
+    // inside writePacket(). In flow-control modes the bounded WriteBuffer
+    // gates reads instead (in discard mode it always has room: old output
+    // is dropped).
+    if (flowControlEnabled ? terminalOutputBuffer.canAcceptMore()
+                           : serverClientState->canBufferWrite(2 * BUF_SIZE)) {
       FD_SET(terminalFd, &rfd);
       maxfd = terminalFd;
     }
     int serverClientFd = serverClientState->getSocketFd();
     if (serverClientFd > 0) {
+      if (flowControlEnabled) {
+        // Reapply every iteration: the connection gets a brand-new socket
+        // on reconnect, kernel tuning is per-socket, and fd numbers are
+        // reused, so there is no reliable way to detect the swap from the
+        // fd alone. The setsockopt is idempotent and costs ~1us.
+        serverSocketHandler->minimizeKernelBuffering(serverClientFd);
+      }
       FD_SET(serverClientFd, &rfd);
       maxfd = max(maxfd, serverClientFd);
+
+      // Wake as soon as the socket can take more buffered output.
+      if (flowControlEnabled && terminalOutputBuffer.hasPendingData()) {
+        FD_SET(serverClientFd, &wfd);
+      }
     }
     // Include port forward sockets in select for low-latency forwarding.
     set<int> pfFds;
@@ -308,9 +424,39 @@ void TerminalServer::runTerminal(
     }
     tv.tv_sec = 0;
     tv.tv_usec = 100000;
-    select(maxfd + 1, &rfd, NULL, NULL, &tv);
+    if (select(maxfd + 1, &rfd, &wfd, NULL, &tv) < 0 && errno == EINTR) {
+      // Interrupted by a signal: the fd sets are unspecified, so
+      // re-evaluate rather than acting on them. Other errors (e.g. a fd
+      // closed by another thread) fall through: the read/write paths then
+      // surface the dead fd as a session-ending error.
+      continue;
+    }
 
     try {
+      // First, drain buffered terminal output while the kernel's unsent
+      // queue is below the low-water mark (TCP_NOTSENT_LOWAT).
+      if (flowControlEnabled && serverClientFd > 0 &&
+          FD_ISSET(serverClientFd, &wfd) &&
+          terminalOutputBuffer.hasPendingData()) {
+        while (terminalOutputBuffer.hasPendingData()) {
+          size_t count;
+          const char* data = terminalOutputBuffer.peekData(&count);
+          if (data == nullptr || count == 0) break;
+
+          et::TerminalBuffer tb;
+          tb.set_buffer(string(data, count));
+          VLOG(2) << "Draining buffered bytes to client: " << count << " "
+                  << serverClientState->getWriter()->getSequenceNumber();
+          serverClientState->writePacket(
+              Packet(TerminalPacketType::TERMINAL_BUFFER, protoToString(tb)));
+          terminalOutputBuffer.consume(count);
+
+          if (!isSocketWritable(serverClientFd)) {
+            break;  // Kernel queue is full enough; stop draining
+          }
+        }
+      }
+
       // Check for data to receive; the received
       // data includes also the data previously sent
       // on the same master descriptor (line 90).
@@ -322,10 +468,15 @@ void TerminalServer::runTerminal(
           VLOG(2) << "Sending bytes from terminal: " << rc << " "
                   << serverClientState->getWriter()->getSequenceNumber();
           string s(b, rc);
-          et::TerminalBuffer tb;
-          tb.set_buffer(s);
-          serverClientState->writePacket(
-              Packet(TerminalPacketType::TERMINAL_BUFFER, protoToString(tb)));
+          if (flowControlEnabled) {
+            // Stage for flow-controlled draining (above)
+            terminalOutputBuffer.enqueue(s);
+          } else {
+            et::TerminalBuffer tb;
+            tb.set_buffer(s);
+            serverClientState->writePacket(
+                Packet(TerminalPacketType::TERMINAL_BUFFER, protoToString(tb)));
+          }
         } else if (rc == 0) {
           LOG(INFO) << "Terminal session ended";
           run = false;
diff --git a/src/terminal/UserJumphostHandler.cpp b/src/terminal/UserJumphostHandler.cpp
index 38c11ca29..167754047 100644
--- a/src/terminal/UserJumphostHandler.cpp
+++ b/src/terminal/UserJumphostHandler.cpp
@@ -40,8 +40,15 @@ void UserJumphostHandler::run() {
   InitialPayload payload;
   while (true) {
     Packet initPacket;
-    if (!routerSocketHandler->readPacket(routerFd, &initPacket)) {
-      continue;
+    try {
+      if (!routerSocketHandler->readPacket(routerFd, &initPacket)) {
+        continue;
+      }
+    } catch (const std::runtime_error& re) {
+      // The router connection died before init (e.g. etserver shut down).
+      // Fail with a logged fatal instead of an uncaught exception.
+      STFATAL << "Router connection died waiting for jumphost init: "
+              << re.what();
     }
     if (initPacket.getHeader() != TerminalPacketType::JUMPHOST_INIT) {
       STFATAL << "Invalid jumphost init packet header: "
diff --git a/src/terminal/UserTerminalHandler.cpp b/src/terminal/UserTerminalHandler.cpp
index b82ad9c3c..011a223a3 100644
--- a/src/terminal/UserTerminalHandler.cpp
+++ b/src/terminal/UserTerminalHandler.cpp
@@ -40,8 +40,15 @@ UserTerminalHandler::UserTerminalHandler(
 void UserTerminalHandler::run() {
   while (true) {
     Packet termInitPacket;
-    if (!socketHandler->readPacket(routerFd, &termInitPacket)) {
-      continue;
+    try {
+      if (!socketHandler->readPacket(routerFd, &termInitPacket)) {
+        continue;
+      }
+    } catch (const std::runtime_error& re) {
+      // The router connection died before init (e.g. etserver shut down).
+      // Fail with a logged fatal instead of an uncaught exception.
+      STFATAL << "Router connection died waiting for terminal init: "
+              << re.what();
     }
     if (termInitPacket.getHeader() != TerminalPacketType::TERMINAL_INIT) {
       STFATAL << "Invalid terminal init packet header: "
@@ -52,6 +59,11 @@ void UserTerminalHandler::run() {
       setenv(ti.environmentnames(a).c_str(), ti.environmentvalues(a).c_str(),
              true);
     }
+    if (ti.flow_control_mode() != et::FLOW_CONTROL_NONE) {
+      // The client opted into flow control: shrink the kernel buffer on the
+      // etterminal->etserver hop so backpressure holds less stale output.
+      socketHandler->minimizeKernelBuffering(routerFd);
+    }
     break;
   }
 
diff --git a/test/e2e/README.md b/test/e2e/README.md
new file mode 100644
index 000000000..d9ac6fae5
--- /dev/null
+++ b/test/e2e/README.md
@@ -0,0 +1,141 @@
+# End-to-End Flow Control Testing
+
+This directory contains tools for testing ET's flow control behavior with a
+simulated slow network.
+
+## Overview
+
+The test uses a TCP throttle proxy between the ET client and server to simulate
+a bandwidth-limited network. The proxy applies TCP backpressure to the server
+(reads slowly), which forces the server's flow control to kick in.
+
+```
+print_timestamps.py -> PTY -> etterminal -> etserver -> [throttle proxy] -> et client -> terminal
+                                                          (100KB/s)
+```
+
+The `--idpasskey` flag on both `et` and `etterminal` bypasses SSH, so you can
+run all three processes on the same machine without SSH access.
+
+## Prerequisites
+
+Build ET first:
+
+```bash
+cd build && cmake .. && ninja
+```
+
+## Quick Start
+
+Run the test script (must run outside process supervisors that kill children,
+e.g. via cron or a dedicated terminal):
+
+```bash
+# From the repo root:
+bash test/e2e/run_e2e_test.sh discard 100000
+```
+
+The script starts etserver (port 4444), a throttle proxy (port 4445 -> 4444 at
+100KB/s), etterminal, and the et client in a tmux session. It then runs
+`print_timestamps.py` and measures:
+
+- **display_lag**: how far behind the timestamps on screen are from real time
+- **process_lag**: how far behind the process producing output is from real time
+  (0 = running freely, >1 = stalled by backpressure)
+
+## Manual Setup
+
+If you prefer to run each component in a separate terminal:
+
+```bash
+# Generate credentials (id must be 16 chars, key must be 32 chars)
+ID="TestID0123456789"
+KEY="TestPasskey0123456789012345678AB"
+
+# Terminal 1: etserver
+./build/etserver --serverfifo=/tmp/test_et.fifo --port=4444
+
+# Terminal 2: throttle proxy (100KB/s)
+python3 test/e2e/throttle_proxy.py 4445 4444 100000
+
+# Terminal 3: etterminal
+./build/etterminal --idpasskey="$ID/$KEY" --serverfifo=/tmp/test_et.fifo
+
+# Terminal 4: et client
+./build/et --idpasskey="$ID/$KEY" 127.0.0.1:4445 --flow-control discard
+
+# Inside the ET session, run:
+python3 test/e2e/print_timestamps.py
+```
+
+Then compare the timestamps on screen with `date` to measure lag.
+
+## What to Expect
+
+### Without flow control (baseline, before this PR)
+
+The server reads from the PTY and writes directly to the TCP socket. When the
+proxy is slow, `writePacket()` blocks, which blocks PTY reads, which stalls the
+process. Data already in the TCP kernel buffer is "old" and gets delivered to the
+client over time.
+
+- **display_lag**: grows linearly (~1s per second)
+- **process_lag**: grows (process stalls)
+- **Ctrl-C**: may be slow because old data must drain first
+
+### With `--flow-control backpressure`
+
+Same as baseline: when the 256KB WriteBuffer fills and the socket is not
+writable, the server stops reading from the PTY. The process stalls.
+
+- **display_lag**: grows (bounded by TCP buffer + WriteBuffer = ~400KB)
+- **process_lag**: grows (process stalls when buffer fills)
+- **Ctrl-C**: responsive because buffered data is bounded
+
+### With `--flow-control discard`
+
+The WriteBuffer always accepts data and discards oldest chunks when full. The
+server keeps reading from the PTY even when the socket is slow. The process
+never stalls. Old data is discarded, so when the socket becomes writable, the
+server sends *recent* data.
+
+- **display_lag**: grows (bounded by TCP buffer, smaller than backpressure)
+- **process_lag**: near zero (process runs freely)
+- **Ctrl-C**: responsive
+
+## Throttle Proxy Details
+
+`throttle_proxy.py` is a TCP proxy that rate-limits the server→client direction
+by reading from the server socket at a controlled rate. This causes the kernel
+TCP send buffer on the server side to fill up, which makes the server's
+`writePacket()` calls block — exactly what happens on a slow network.
+
+The client→server direction (keystrokes, Ctrl-C) is forwarded at full speed.
+
+The proxy binds on `[::]` with `IPV6_V6ONLY=0` (dual-stack) to accept both IPv4
+and IPv6-mapped IPv4 connections, which is required because ET's
+`TcpSocketHandler::connect()` may use either address family depending on
+`getaddrinfo()` results.
+
+## Running via Cron (for automated environments)
+
+If your environment kills background processes (e.g. Claude Code), schedule the
+test via cron:
+
+```bash
+echo "* * * * * bash /path/to/test/e2e/run_e2e_test.sh discard 100000 > /tmp/e2e_result.log 2>&1" | crontab -
+# Wait ~2 minutes, then:
+cat /tmp/e2e_result.log
+crontab -r
+```
+
+## Troubleshooting
+
+- **"proxy saw no connection"**: ET connected to a different port (likely the
+  system etserver on port 2022). Make sure `--idpasskey` is being used so SSH
+  config parsing is skipped.
+- **etserver crashes with EINVAL**: Fixed by handling EBADF/EINVAL in
+  `waitOnSocketWritable()` (see Headers.hpp). The fd becomes invalid when the
+  client disconnects mid-drain.
+- **"Connection refused" on proxy**: The proxy process died. Check if port 4445
+  is already in use (`lsof -i:4445`).
diff --git a/test/e2e/REPORT.md b/test/e2e/REPORT.md
new file mode 100644
index 000000000..4fb9caf19
--- /dev/null
+++ b/test/e2e/REPORT.md
@@ -0,0 +1,165 @@
+# ET Flow Control: E2E Test Report
+
+## Problem Statement (Issue #631)
+
+When a process inside an ET session produces output faster than the network can
+deliver it, ET has no mechanism to manage the mismatch. The result:
+
+1. The terminal display falls progressively behind real time
+2. Ctrl-C appears to do nothing: the interrupt is delivered, but tens of
+   seconds of stale queued output must drain before the prompt reappears
+3. In the worst case (laptop sleep, wifi drop), long-running jobs freeze
+   entirely because the PTY kernel buffer fills up and `write()` blocks
+
+## The fix, in two parts
+
+**Part 1: application-level flow control.** Terminal output is staged in a
+bounded `WriteBuffer` between the PTY and the client socket. Flow control
+is strictly opt-in; users who don't opt in are completely unchanged:
+
+- `--flow-control none` (default): the exact pre-feature code path. Output
+  is written straight to the connection with blocking writes; no
+  application buffering, no kernel socket tuning.
+- `--flow-control backpressure`: when the buffer fills, stop reading from
+  the PTY. The remote process pauses, exactly like plain ssh. Lossless —
+  required for correctness-sensitive consumers like `tmux -CC`.
+- `--flow-control discard`: when the buffer fills, drop the *oldest* pending
+  output. The remote process never stalls and the display stays close to
+  real time. Old output is lost from scrollback while the link is saturated.
+
+**Part 2: kernel buffer tuning (opt-in sessions only).** Flow control only
+helps if pending data actually waits in the application buffer. An audit of
+every buffer on the path found kernel-side reservoirs that were absorbing
+the backlog downstream of any point where ET could drop or gate it. These
+are tuned per-session, only when the client opted in (the server learns the
+mode from InitialPayload; etterminal learns it via TermInit):
+
+| Buffer | Untuned size | Fix |
+|---|---|---|
+| TCP send buffer (server->client) | autotunes to multi-MB (20MB max on this host) | `TCP_NOTSENT_LOWAT=32KB`: select() only reports writable when <32KB is unsent, so the backlog stays in the WriteBuffer. In-flight (sent-but-unacked) data is not limited, preserving high-BDP throughput. |
+| ET WriteBuffer | 256KB | 64KB — only needs to absorb bursts between drain opportunities |
+| etterminal->etserver unix socket | ~200KB (net.core.wmem_default) | `SO_SNDBUF=64KB` on the sender side |
+| PTY kernel buffer | ~64KB | not tunable from userspace; part of the fixed lag floor |
+| BackedWriter backup (64MB) | n/a | reconnect recovery only; never delays live data |
+
+## Test Setup
+
+```
+print_timestamps.py -> PTY -> etterminal -> etserver --[TCP]--> throttle_proxy --[TCP]--> et client -> terminal
+                                                       100KB/s
+```
+
+All components run on a single machine via `--idpasskey` (no SSH). A userspace
+TCP throttle proxy between etserver and the et client limits server-to-client
+throughput to 100KB/s. The proxy clamps `SO_RCVBUF` on its server-facing
+socket to 64KB so it models a real slow link instead of silently absorbing
+megabytes in its own autotuned receive buffer.
+
+**Workload**: `print_timestamps.py` prints 1000 timestamped lines in a burst
+every 10ms (~2.7MB/s raw output — 27x the proxy's capacity).
+
+**Metrics** (sampled every 5s; Ctrl-C sent after the workload):
+
+- **display_lag**: wall clock minus the newest timestamp visible on the
+  client's terminal — how stale the screen is.
+- **process_lag**: wall clock minus the timestamp the producer most recently
+  wrote (via a sidecar file that bypasses ET). Large values mean the process
+  is blocked in `write()`.
+- **ctrl_c_latency**: time from sending Ctrl-C until the shell prompt is
+  visible again. This is the original issue 631 complaint.
+
+## Results (2026-07-10, devvm, Linux 6.13; re-validated on the v7.0.0 base)
+
+Upstream v7.0.0 independently added disconnect buffering (#731, #762: the
+connection buffers up to 64MB while disconnected so processes don't freeze
+immediately) and a PTY-input deadlock fix (#765). Those address the
+disconnect freeze, but not the connected saturated link: with a live but
+slow connection, writes still land in the autotuned kernel send buffer, so
+the `none` baseline below reproduces issue #631 unchanged on v7.0.0.
+
+### Slow link, 30s saturation
+
+| scenario | display lag @30s | process | Ctrl-C -> prompt |
+|---|---|---|---|
+| `none` (default — status quo, v7.0.0 base) | 32.3s, growing ~1s/s | throttled to link rate | **115.8s** |
+| backpressure, no kernel tuning (pre-v7 base) | 31.5s, growing | throttled | 53.8s |
+| discard, no kernel tuning (pre-v7 base) | 24.4s, growing | full speed | 28.0s |
+| **backpressure + tuning (opt-in)** | **~2s, bounded** | throttled (lossless) | **2.7s** |
+| **discard + tuning (opt-in)** | **~1.1s, bounded** | **full speed** | **2.1s** |
+
+The `none` row is the default build with no flag: unbounded kernel queue,
+growing lag, and Ctrl-C takes however long the queue takes to drain (high
+variance run to run; 56-116s observed across bases). This is deliberate:
+current users see zero change unless they opt in.
+
+The no-kernel-tuning rows show why the tuning is essential: with the default
+autotuned TCP send buffer, megabytes of stale output pile up in the kernel
+where the WriteBuffer can neither gate nor drop them, and both modes are
+barely better than trunk. With `TCP_NOTSENT_LOWAT`, display lag stops growing
+entirely: it is bounded by the small fixed pipeline (WriteBuffer + unsent
+bytes + link queue) instead of scaling with how long the link has been
+saturated.
+
+### 60s disconnect at t=30s (tuned)
+
+| scenario | process during disconnect | after reconnect | Ctrl-C |
+|---|---|---|---|
+| discard | **full speed, never stalls** | fresh output in <5s, no stale replay | 1.6-2.1s |
+| backpressure | stalls (lossless contract, like ssh; the v7 default instead buffers up to 64MB before stalling) | resumes, display recovers in <5s | 2.6s |
+
+etserver stayed alive through all scenarios.
+
+### Bulk throughput (localhost, no throttle, 30MB through the full pipeline)
+
+Measurements are noisy on a shared devvm (trunk build: 40-47 MB/s across
+runs; this branch: 20-41 MB/s across modes, with `none`, `backpressure`,
+and `discard` overlapping run-to-run). Two things hold: all modes stay in
+the tens of MB/s, far beyond any realistic terminal workload, and the
+default `none` path is the identical pre-feature code, so non-opted-in
+users cannot regress. On real WAN links throughput is BDP-bound, which
+`TCP_NOTSENT_LOWAT` does not restrict.
+
+## Why the default is `none`
+
+The requirement is that current users are no worse off — the strongest
+form of that is byte-identical behavior, so the default (CLI flag and
+proto field alike, which also covers old clients talking to new servers)
+is the legacy path with stock kernel buffering.
+
+`backpressure` is the conservative opt-in: it keeps today's lossless
+semantics (a slow client pauses the producer, exactly like plain ssh; safe
+for `tmux -CC`) while bounding the queue so Ctrl-C takes ~3s instead of
+minutes. `discard` is the opt-in for freshness over completeness — the
+`cat /dev/zero | base64` case from the issue, ML training jobs, builds:
+the process never stalls (even fully disconnected) and the display tracks
+real time. The cost is that saturated-link output is missing from
+scrollback.
+
+## What this means for real-world use
+
+- **ML training / long build, laptop closed**: with `none` or
+  `backpressure`, the job pauses when buffers fill — same as today. With
+  `--flow-control discard`, the job keeps running at full speed and you see
+  current output when you reconnect.
+- **`cat` a huge file over a slow link, then Ctrl-C**: prompt returns in
+  ~2-3s in either opt-in mode (default/today: a minute or more).
+- **tmux -CC**: use `none` (default) or `backpressure`. Every byte is
+  delivered.
+
+## Reproducing These Results
+
+```bash
+cd /path/to/EternalTerminal/build
+cmake -DDISABLE_VCPKG=ON -GNinja .. && ninja -j4
+
+# One scenario (trunk|none|backpressure|discard) [--disconnect]
+bash test/e2e/do_scenario.sh discard --outdir /tmp/et_e2e
+bash test/e2e/do_scenario.sh discard --disconnect --outdir /tmp/et_e2e
+
+# Bulk throughput on a fast link
+bash test/e2e/throughput_test.sh
+bash test/e2e/throughput_test.sh --flow-control discard
+```
+
+See `test/e2e/README.md` for manual setup and `test/e2e/throttle_proxy.py` for
+details on the proxy implementation.
diff --git a/test/e2e/do_scenario.sh b/test/e2e/do_scenario.sh
new file mode 100644
index 000000000..e86bc8083
--- /dev/null
+++ b/test/e2e/do_scenario.sh
@@ -0,0 +1,285 @@
+#!/bin/bash
+# Runs one E2E flow control scenario. Outputs JSONL metrics.
+#
+# Usage: bash do_scenario.sh <mode> [--disconnect] [--outdir DIR]
+#   mode: trunk | backpressure | discard
+#   --disconnect: simulate 60s TCP disconnect at t=30s
+#   --outdir: directory for JSONL output and logs
+#
+# Commits:
+#   trunk:        ee8ddc21c (base) + f8930e169 (harness: --idpasskey, crash fix)
+#   backpressure: 497000e08 (--flow-control backpressure)
+#   discard:      497000e08 (--flow-control discard)
+
+MODE="$1"; shift
+DISCONNECT=false
+OUTDIR="/tmp/et_e2e_results"
+while [ $# -gt 0 ]; do
+    case "$1" in
+        --disconnect) DISCONNECT=true ;;
+        --outdir) OUTDIR="$2"; shift ;;
+    esac
+    shift
+done
+
+DIR="$(cd "$(dirname "$0")" && pwd)"
+REPO="$(cd "$DIR/../.." && pwd)"
+BUILD="$REPO/build"
+RATE=100000
+ID="E2E$(date +%s | tail -c 8)"
+KEY="E2ETestKey123456789012345678901A"
+SIDECAR="/tmp/et_e2e_sidecar_$$.txt"
+ETMUX="/tmp/et_e2e_client_$$.sock"
+PROXY_LOG="/tmp/et_e2e_proxy_$$.log"
+
+# The cleanup trap restores src/, proto/, and test dirs to HEAD (the trunk
+# scenario checks out old sources there). Any uncommitted work in those
+# paths would be silently destroyed, so refuse to run on a dirty tree.
+if ! git -C "$REPO" diff --quiet -- src/ proto/ test/integration_tests/ test/unit_tests/ 2>/dev/null; then
+    echo "ERROR: uncommitted changes under src/, proto/, or test/. Commit them first:"
+    git -C "$REPO" status --short -- src/ proto/ test/integration_tests/ test/unit_tests/ | head
+    exit 1
+fi
+
+mkdir -p "$OUTDIR"
+
+SCENARIO="$MODE"
+[ "$DISCONNECT" = "true" ] && SCENARIO="${MODE}-disconnect"
+JSONL="$OUTDIR/${SCENARIO}.jsonl"
+> "$JSONL"
+
+C='\033[1;36m'; G='\033[1;32m'; R='\033[1;31m'; Y='\033[33m'; N='\033[0m'
+
+cleanup() {
+    tmux -S "$ETMUX" kill-server 2>/dev/null
+    kill $ETSERVER_PID $PROXY_PID 2>/dev/null
+    kill -9 $(lsof -t -i:4444 2>/dev/null) $(lsof -t -i:4445 2>/dev/null) 2>/dev/null
+    pkill -9 -f "etterminal.*$ID" 2>/dev/null
+    rm -f /tmp/et_e2e_demo.fifo "$ETMUX" "$SIDECAR" "$PROXY_LOG"
+    # Restore only the dirs the trunk scenario checks out. Do NOT use
+    # "git checkout HEAD -- ." here: it clobbers unrelated uncommitted work
+    # in the tree (this exact mistake once reverted the flow-control
+    # implementation itself).
+    cd "$REPO" && git checkout HEAD -- src/ proto/ test/integration_tests/ test/unit_tests/ 2>/dev/null
+}
+trap cleanup EXIT
+
+# --- Checkout and build ---
+
+COMMIT_DESC=""
+ET_CMD=""
+case "$MODE" in
+    trunk)
+        echo -e "${C}=== Scenario: trunk (no flow control) ===${N}"
+        COMMIT_DESC="ee8ddc21c + f8930e169 (harness only, no WriteBuffer)"
+        ET_CMD="./et --idpasskey=ID/KEY 127.0.0.1:4445"
+        cd "$REPO"
+        git checkout ee8ddc21c -- src/ proto/ test/integration_tests/ test/unit_tests/ 2>/dev/null
+        git checkout f8930e169 -- src/terminal/TerminalClientMain.cpp src/terminal/TerminalMain.cpp src/base/Headers.hpp 2>/dev/null
+        rm -f src/base/WriteBuffer.hpp test/unit_tests/WriteBufferTest.cpp
+        ;;
+    none)
+        echo -e "${C}=== Scenario: none (HEAD, no --flow-control: must match trunk) ===${N}"
+        COMMIT_DESC="HEAD ($(cd "$REPO" && git rev-parse --short HEAD)), no flag"
+        ET_CMD="./et --idpasskey=ID/KEY 127.0.0.1:4445"
+        # Use HEAD as-is (no checkout needed)
+        ;;
+    backpressure)
+        echo -e "${C}=== Scenario: backpressure ===${N}"
+        COMMIT_DESC="HEAD ($(cd "$REPO" && git rev-parse --short HEAD))"
+        ET_CMD="./et --idpasskey=ID/KEY --flow-control backpressure 127.0.0.1:4445"
+        # Use HEAD as-is (no checkout needed)
+        ;;
+    discard)
+        echo -e "${C}=== Scenario: discard ===${N}"
+        COMMIT_DESC="HEAD ($(cd "$REPO" && git rev-parse --short HEAD))"
+        ET_CMD="./et --idpasskey=ID/KEY --flow-control discard 127.0.0.1:4445"
+        # Use HEAD as-is (no checkout needed)
+        ;;
+    *) echo "Usage: $0 <trunk|none|backpressure|discard> [--disconnect]"; exit 1 ;;
+esac
+FC_FLAG=""
+[ "$MODE" = "backpressure" ] && FC_FLAG="--flow-control backpressure"
+[ "$MODE" = "discard" ] && FC_FLAG="--flow-control discard"
+
+echo "  commit: $COMMIT_DESC"
+echo "  client: $ET_CMD"
+echo "  proxy:  ${RATE}B/s throttle on port 4445"
+[ "$DISCONNECT" = "true" ] && echo "  disconnect: 60s at t=30s"
+echo ""
+
+echo -e "${C}--- Build ---${N}"
+cd "$BUILD"
+cmake -DDISABLE_VCPKG=ON -GNinja .. 2>&1 | tail -1
+ninja -j4 2>&1 | tail -3
+echo ""
+
+# --- Start services ---
+
+echo -e "${C}--- Services ---${N}"
+rm -f /tmp/et_e2e_demo.fifo
+
+"$BUILD/etserver" --serverfifo=/tmp/et_e2e_demo.fifo --port=4444 --logdir="$OUTDIR" &>/dev/null &
+ETSERVER_PID=$!
+sleep 3
+
+PYTHONUNBUFFERED=1 python3 -u "$DIR/throttle_proxy.py" 4445 4444 "$RATE" >"$PROXY_LOG" 2>&1 &
+PROXY_PID=$!
+sleep 2
+
+"$BUILD/etterminal" --idpasskey="$ID/$KEY" --serverfifo=/tmp/et_e2e_demo.fifo --logdir="$OUTDIR" &>/dev/null &
+sleep 3
+
+for i in $(seq 1 15); do
+    ES=$(lsof -i:4444 2>/dev/null | grep -c LISTEN)
+    PX=$(lsof -i:4445 2>/dev/null | grep -c LISTEN)
+    [ "$ES" -ge 1 ] && [ "$PX" -ge 1 ] && break
+    sleep 1
+done
+[ "$ES" -lt 1 ] || [ "$PX" -lt 1 ] && { echo -e "${R}FAILED: etserver=$ES proxy=$PX${N}"; exit 1; }
+echo -e "${G}etserver=:4444 proxy=:4445 (${RATE}B/s)${N}"
+
+# --- Connect ---
+
+echo -e "${C}--- Connect ---${N}"
+tmux -S "$ETMUX" new-session -d -s et -x 200 -y 50
+for i in $(seq 1 30); do
+    tmux -S "$ETMUX" capture-pane -p 2>/dev/null | grep -q '\$' && break
+    sleep 1
+done
+tmux -S "$ETMUX" send-keys "$BUILD/et --idpasskey='$ID/$KEY' 127.0.0.1:4445 $FC_FLAG" Enter
+
+for i in $(seq 1 30); do
+    grep -q connect "$PROXY_LOG" 2>/dev/null && break
+    sleep 1
+    [ "$i" -eq 30 ] && { echo -e "${R}FAILED: no proxy connection${N}"; cat "$PROXY_LOG"; exit 1; }
+done
+echo -e "${G}Connected through proxy${N}"
+
+for i in $(seq 1 30); do
+    tmux -S "$ETMUX" capture-pane -p 2>/dev/null | grep -qF '~]$' && break
+    sleep 1
+done
+echo -e "${G}Remote shell ready${N}"
+echo ""
+
+# --- Run workload ---
+
+TOTAL_DURATION=30
+[ "$DISCONNECT" = "true" ] && TOTAL_DURATION=120
+DISCONNECT_AT=30
+RECONNECT_AT=90
+
+echo -e "${C}--- print_timestamps.py (${TOTAL_DURATION}s) ---${N}"
+tmux -S "$ETMUX" send-keys "SIDECAR_FILE=$SIDECAR python3 $DIR/print_timestamps.py" Enter
+
+PREV_LINES=0
+PREV_ELAPSED="0"
+
+printf "%-6s  %-12s  %-12s  %-12s  %-10s  %-10s\n" "TIME" "DISPLAY_LAG" "PROCESS_LAG" "LINES/SEC" "CONNECTED" "STATUS"
+printf "%-6s  %-12s  %-12s  %-12s  %-10s  %-10s\n" "----" "-----------" "-----------" "---------" "---------" "------"
+
+ELAPSED=0
+while [ "$ELAPSED" -lt "$TOTAL_DURATION" ]; do
+    sleep 5
+    ELAPSED=$((ELAPSED + 5))
+
+    if [ "$DISCONNECT" = "true" ]; then
+        [ "$ELAPSED" -eq "$DISCONNECT_AT" ] && { echo -e "${R}>>> DISCONNECT <<<${N}"; kill -USR1 $PROXY_PID 2>/dev/null; }
+        [ "$ELAPSED" -eq "$RECONNECT_AT" ] && { echo -e "${G}>>> RECONNECT <<<${N}"; kill -USR2 $PROXY_PID 2>/dev/null; }
+    fi
+
+    CONNECTED="true"
+    [ "$DISCONNECT" = "true" ] && [ "$ELAPSED" -gt "$DISCONNECT_AT" ] && [ "$ELAPSED" -le "$RECONNECT_AT" ] && CONNECTED="false"
+
+    SCREEN=$(tmux -S "$ETMUX" capture-pane -p 2>/dev/null | grep -oE '[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]+' | tail -1)
+    REAL=$(python3 -c "from datetime import datetime; print(datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'))")
+
+    DLAG="null"
+    [ -n "$SCREEN" ] && DLAG=$(python3 -c "
+from datetime import datetime
+s=datetime.strptime('${SCREEN}'.strip(),'%Y-%m-%d %H:%M:%S.%f')
+r=datetime.strptime('${REAL}'.strip(),'%Y-%m-%d %H:%M:%S.%f')
+print(f'{(r-s).total_seconds():.1f}')" 2>/dev/null || true)
+
+    PLAG="null"; LINES_SEC="null"; CUR_LINES=0; CUR_ELAPSED="0"
+    if [ -f "$SIDECAR" ]; then
+        IFS=',' read -r PROCESS_TS CUR_LINES CUR_ELAPSED < "$SIDECAR"
+        [ -n "$PROCESS_TS" ] && PLAG=$(python3 -c "
+from datetime import datetime
+s=datetime.strptime('${PROCESS_TS}'.strip(),'%Y-%m-%d %H:%M:%S.%f')
+r=datetime.strptime('${REAL}'.strip(),'%Y-%m-%d %H:%M:%S.%f')
+print(f'{(r-s).total_seconds():.1f}')" 2>/dev/null || true)
+        [ -n "$CUR_ELAPSED" ] && [ -n "$CUR_LINES" ] && LINES_SEC=$(python3 -c "
+cl=int('${CUR_LINES}'); pl=int('${PREV_LINES}')
+ct=float('${CUR_ELAPSED}'); pt=float('${PREV_ELAPSED}')
+dt=ct-pt
+print(f'{(cl-pl)/dt:.0f}' if dt > 0.1 else '0')" 2>/dev/null || true)
+    fi
+    PREV_LINES="${CUR_LINES:-0}"; PREV_ELAPSED="${CUR_ELAPSED:-0}"
+
+    PSTATUS="running"
+    [ "$PLAG" != "null" ] && ! python3 -c "exit(0 if float('$PLAG') < 1.0 else 1)" 2>/dev/null && PSTATUS="STALLED"
+
+    DLAG_S="${DLAG}s"; [ "$DLAG" = "null" ] && DLAG_S="?"
+    PLAG_S="${PLAG}s"; [ "$PLAG" = "null" ] && PLAG_S="?"
+    LS_S="$LINES_SEC"; [ "$LINES_SEC" = "null" ] && LS_S="?"
+
+    COLOR="$G"; [ "$PSTATUS" = "STALLED" ] && COLOR="$R"
+    printf "t=%-3ss  %-12s  %-12s  %-12s  %-10s  ${COLOR}%-10s${N}\n" "$ELAPSED" "$DLAG_S" "$PLAG_S" "$LS_S" "$CONNECTED" "$PSTATUS"
+
+    echo "{\"t\":$ELAPSED,\"display_lag\":$DLAG,\"process_lag\":$PLAG,\"lines_sec\":$LINES_SEC,\"connected\":$CONNECTED}" >> "$JSONL"
+done
+
+# --- Ctrl-C responsiveness ---
+# This is the original issue 631 complaint: with a saturated link, ^C takes
+# a very long time to visibly take effect because stale queued output must
+# drain before the prompt reappears.
+echo ""
+echo -e "${C}--- Ctrl-C responsiveness ---${N}"
+CC_T0=$(python3 -c 'import time; print(f"{time.time():.3f}")')
+tmux -S "$ETMUX" send-keys C-c
+CTRL_C_LATENCY="null"
+for i in $(seq 1 240); do
+    sleep 0.5
+    LAST_LINES=$(tmux -S "$ETMUX" capture-pane -p 2>/dev/null | sed '/^$/d' | tail -2)
+    if echo "$LAST_LINES" | grep -qF '~]$'; then
+        CTRL_C_LATENCY=$(python3 -c "import time; print(f'{time.time() - $CC_T0:.1f}')")
+        break
+    fi
+done
+if [ "$CTRL_C_LATENCY" = "null" ]; then
+    echo -e "${R}Ctrl-C: prompt did not return within 120s${N}"
+else
+    echo -e "Ctrl-C to prompt: ${G}${CTRL_C_LATENCY}s${N}"
+fi
+echo "{\"event\":\"ctrl_c\",\"latency\":$CTRL_C_LATENCY}" >> "$JSONL"
+
+echo ""
+ES_ALIVE=$(lsof -i:4444 2>/dev/null | grep -c LISTEN)
+[ "$ES_ALIVE" -gt 0 ] && echo -e "etserver: ${G}alive${N}" || echo -e "etserver: ${R}CRASHED${N}"
+
+cp "$PROXY_LOG" "$OUTDIR/${SCENARIO}_proxy.log" 2>/dev/null
+
+python3 -c "
+import json
+meta = {
+    'scenario': '$SCENARIO',
+    'mode': '$MODE',
+    'disconnect': $( [ \"$DISCONNECT\" = \"true\" ] && echo True || echo False ),
+    'commit': '$COMMIT_DESC',
+    'et_command': '$ET_CMD',
+    'proxy_rate': $RATE,
+    'ctrl_c_latency': $CTRL_C_LATENCY,
+    'etserver_alive': $( [ "$ES_ALIVE" -gt 0 ] && echo True || echo False ),
+}
+with open('$OUTDIR/${SCENARIO}_meta.json', 'w') as f:
+    json.dump(meta, f, indent=2)
+"
+
+echo "Results: $JSONL"
+
+tmux -S "$ETMUX" send-keys C-c 2>/dev/null
+sleep 1
+tmux -S "$ETMUX" send-keys "exit" Enter 2>/dev/null
+sleep 1
diff --git a/test/e2e/generate_report.py b/test/e2e/generate_report.py
new file mode 100644
index 000000000..6cd0bb2c4
--- /dev/null
+++ b/test/e2e/generate_report.py
@@ -0,0 +1,194 @@
+#!/usr/bin/env python3
+"""Generate a self-contained HTML report with inline SVG charts from E2E JSONL data.
+
+Usage: python3 generate_report.py <results_dir> <output.html>
+"""
+import json
+import sys
+from datetime import datetime
+from pathlib import Path
+
+
+def load_scenario(results_dir, name):
+    jsonl = Path(results_dir) / f"{name}.jsonl"
+    meta_file = Path(results_dir) / f"{name}_meta.json"
+    if not jsonl.exists():
+        return None
+    samples = []
+    for line in jsonl.read_text().strip().split("\n"):
+        if line.strip():
+            samples.append(json.loads(line))
+    meta = json.loads(meta_file.read_text()) if meta_file.exists() else {}
+    return {"name": name, "samples": samples, "meta": meta}
+
+
+def svg_chart(scenarios, key, title, ylabel, width=700, height=300, events=None):
+    colors = {
+        "trunk": "#e74c3c", "backpressure": "#f39c12",
+        "discard": "#2ecc71", "discard-disconnect": "#3498db",
+        "trunk-disconnect": "#c0392b", "backpressure-disconnect": "#e67e22",
+    }
+    ml, mr, mt, mb = 70, 140, 40, 50
+    pw, ph = width - ml - mr, height - mt - mb
+
+    all_t, all_v = [], []
+    for sc in scenarios:
+        for s in sc["samples"]:
+            all_t.append(s["t"])
+            v = s.get(key)
+            if v is not None and str(v) != "null":
+                try:
+                    all_v.append(float(v))
+                except (ValueError, TypeError):
+                    pass
+    if not all_v:
+        return f"<p>No data for {title}</p>"
+
+    t_min, t_max = min(all_t), max(all_t)
+    v_max = max(all_v) * 1.1 or 1
+
+    def tx(t): return ml + (t - t_min) / max(t_max - t_min, 1) * pw
+    def ty(v): return mt + ph - v / v_max * ph
+
+    out = [f'<svg width="{width}" height="{height}" xmlns="http://www.w3.org/2000/svg">',
+           f'<rect width="{width}" height="{height}" fill="white"/>',
+           f'<text x="{width//2}" y="20" text-anchor="middle" font-size="14" font-weight="bold" fill="#333">{title}</text>']
+
+    for i in range(6):
+        y = mt + i * ph / 5
+        v = v_max * (5 - i) / 5
+        out.append(f'<line x1="{ml}" y1="{y}" x2="{ml+pw}" y2="{y}" stroke="#eee"/>')
+        out.append(f'<text x="{ml-5}" y="{y+4}" text-anchor="end" font-size="10" fill="#666">{v:.1f}</text>')
+
+    step = max(int((t_max - t_min) / 8), 5)
+    for t in range(0, int(t_max) + 1, step):
+        x = tx(t)
+        out.append(f'<line x1="{x}" y1="{mt+ph}" x2="{x}" y2="{mt+ph+5}" stroke="#666"/>')
+        out.append(f'<text x="{x}" y="{mt+ph+18}" text-anchor="middle" font-size="10" fill="#666">{t}s</text>')
+
+    out.append(f'<line x1="{ml}" y1="{mt}" x2="{ml}" y2="{mt+ph}" stroke="#333"/>')
+    out.append(f'<line x1="{ml}" y1="{mt+ph}" x2="{ml+pw}" y2="{mt+ph}" stroke="#333"/>')
+    out.append(f'<text x="{width//2}" y="{height-5}" text-anchor="middle" font-size="11" fill="#333">Time (seconds)</text>')
+    out.append(f'<text x="15" y="{height//2}" text-anchor="middle" font-size="11" fill="#333" transform="rotate(-90 15 {height//2})">{ylabel}</text>')
+
+    if events:
+        for evt_t, lbl, clr in events:
+            x = tx(evt_t)
+            out.append(f'<line x1="{x}" y1="{mt}" x2="{x}" y2="{mt+ph}" stroke="{clr}" stroke-width="2" stroke-dasharray="5,5"/>')
+            out.append(f'<text x="{x+3}" y="{mt+12}" font-size="9" fill="{clr}">{lbl}</text>')
+
+    ly = mt + 10
+    for sc in scenarios:
+        color = colors.get(sc["name"], "#999")
+        pts = [(s["t"], float(s[key])) for s in sc["samples"]
+               if s.get(key) is not None and str(s[key]) != "null"
+               and _safe_float(s[key]) is not None]
+        if not pts:
+            continue
+        path = " ".join(f"{'M' if i==0 else 'L'} {tx(t):.1f} {ty(v):.1f}" for i,(t,v) in enumerate(pts))
+        out.append(f'<path d="{path}" fill="none" stroke="{color}" stroke-width="2"/>')
+        for t, v in pts:
+            out.append(f'<circle cx="{tx(t):.1f}" cy="{ty(v):.1f}" r="3" fill="{color}"/>')
+        lx = ml + pw + 10
+        out.append(f'<line x1="{lx}" y1="{ly}" x2="{lx+20}" y2="{ly}" stroke="{color}" stroke-width="2"/>')
+        out.append(f'<text x="{lx+25}" y="{ly+4}" font-size="10" fill="#333">{sc["name"]}</text>')
+        ly += 18
+
+    out.append("</svg>")
+    return "\n".join(out)
+
+
+def _safe_float(v):
+    try:
+        return float(v)
+    except (ValueError, TypeError):
+        return None
+
+
+def generate_html(results_dir, output_path):
+    scenarios = [s for s in [load_scenario(results_dir, n)
+                 for n in ["trunk", "backpressure", "discard",
+                           "trunk-disconnect", "backpressure-disconnect",
+                           "discard-disconnect"]] if s]
+    if not scenarios:
+        print(f"No data in {results_dir}")
+        sys.exit(1)
+
+    events = [(30, "disconnect", "#e74c3c"), (90, "reconnect", "#2ecc71")]
+    has_dc = any(s["name"] == "discard-disconnect" for s in scenarios)
+
+    h = ['<!DOCTYPE html><html><head><meta charset="utf-8">',
+         '<title>ET Flow Control E2E Report</title>',
+         '''<style>
+body{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',sans-serif;max-width:900px;margin:40px auto;padding:0 20px;color:#333;line-height:1.6}
+h1{border-bottom:2px solid #2ecc71;padding-bottom:10px}
+h2{color:#2c3e50;margin-top:40px}
+table{border-collapse:collapse;width:100%;margin:20px 0}
+th,td{border:1px solid #ddd;padding:8px 12px;text-align:left}
+th{background:#f5f5f5}
+.chart{margin:30px 0;text-align:center}
+.good{color:#2ecc71;font-weight:bold}
+.bad{color:#e74c3c;font-weight:bold}
+code{background:#f5f5f5;padding:2px 6px;border-radius:3px;font-size:13px}
+.meta{background:#f9f9f9;border-left:4px solid #3498db;padding:15px;margin:20px 0}
+</style>''',
+         '</head><body>',
+         '<h1>ET Flow Control E2E Test Report</h1>',
+         f'<p>Generated: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}</p>',
+         '<h2>Test Scenarios</h2>',
+         '<table><tr><th>Scenario</th><th>Git Commit</th><th>Client Command</th><th>Proxy Rate</th><th>Server</th></tr>']
+
+    for sc in scenarios:
+        m = sc["meta"]
+        alive = '<span class="good">alive</span>' if m.get("etserver_alive", True) else '<span class="bad">CRASHED</span>'
+        rate = m.get("proxy_rate", "?")
+        rate_str = f"{rate:,} B/s" if isinstance(rate, int) else str(rate)
+        h.append(f'<tr><td><b>{sc["name"]}</b></td><td><code>{m.get("commit","?")}</code></td>'
+                 f'<td><code>{m.get("et_command","?")}</code></td>'
+                 f'<td>{rate_str}</td><td>{alive}</td></tr>')
+    h.append('</table>')
+
+    h.append('<div class="meta"><b>Setup:</b> etserver :4444 &rarr; throttle proxy :4445 &rarr; et client. '
+             'Workload: <code>print_timestamps.py</code> ~2.7MB/s. Proxy limits to 100KB/s.</div>')
+
+    h.append('<h2>Display Lag</h2>')
+    h.append('<p>How stale the terminal output is. Grows ~1s/s in all modes (TCP buffer bottleneck).</p>')
+    h.append(f'<div class="chart">{svg_chart(scenarios, "display_lag", "Display Lag Over Time", "Seconds behind real time", events=events if has_dc else None)}</div>')
+
+    h.append('<h2>Process Lag</h2>')
+    h.append('<p>Whether the producing process is stalled. <span class="good">0 = running freely</span>, <span class="bad">&gt;1 = blocked on write()</span>.</p>')
+    h.append(f'<div class="chart">{svg_chart(scenarios, "process_lag", "Process Lag Over Time", "Seconds behind real time", events=events if has_dc else None)}</div>')
+
+    h.append('<h2>Process Throughput</h2>')
+    h.append('<p>Lines/sec written by the producing process. Higher = process running freely.</p>')
+    h.append(f'<div class="chart">{svg_chart(scenarios, "lines_sec", "Process Throughput", "Lines/sec", events=events if has_dc else None)}</div>')
+
+    h.append('<h2>Interpretation</h2>')
+    h.append('<h3>Display lag is a network property</h3>')
+    h.append('<p>All modes show ~1s/s growth because the proxy limits throughput. TCP is ordered &mdash; '
+             'data in the kernel buffer must be delivered in sequence. No application-level change can fix this.</p>')
+    h.append('<h3>Process lag is the key differentiator</h3>')
+    h.append('<p><b>Trunk &amp; backpressure:</b> process stalls intermittently (process_lag &gt; 0) when TCP buffer fills.</p>')
+    h.append('<p><b>Discard:</b> process_lag stays at <span class="good">0</span>. Old data dropped, PTY always drained.</p>')
+
+    if has_dc:
+        h.append('<h3>Disconnect scenario</h3>')
+        h.append('<p>At t=30s, proxy kills TCP connections for 60s. At t=90s, proxy resumes.</p>')
+        h.append('<p>In <b>discard</b> mode, the process keeps running during disconnect. '
+                 'WriteBuffer discards old output. On reconnect, client sees recent data.</p>')
+
+    h.append('<h3>When to use each mode</h3>')
+    h.append('<table><tr><th>Mode</th><th>Best for</th><th>Trade-off</th></tr>')
+    h.append('<tr><td>discard (default)</td><td>Long-running jobs, builds</td><td>Old output lost</td></tr>')
+    h.append('<tr><td>backpressure</td><td>tmux -CC, stateful protocols</td><td>Process stalls</td></tr></table>')
+    h.append('</body></html>')
+
+    Path(output_path).write_text("\n".join(h))
+    print(f"Report: {output_path}")
+
+
+if __name__ == "__main__":
+    if len(sys.argv) < 3:
+        print("Usage: python3 generate_report.py <results_dir> <output.html>")
+        sys.exit(1)
+    generate_html(sys.argv[1], sys.argv[2])
diff --git a/test/e2e/print_timestamps.py b/test/e2e/print_timestamps.py
new file mode 100644
index 000000000..b047f0241
--- /dev/null
+++ b/test/e2e/print_timestamps.py
@@ -0,0 +1,27 @@
+"""Prints timestamped output at high volume to test flow control.
+
+Produces ~2.7MB/s of output (1000 lines per burst, bursts every 10ms).
+
+If SIDECAR_FILE is set, writes the current timestamp and cumulative line
+count to that file after each burst, enabling external measurement of:
+  - process_lag: is the process keeping up with wall clock?
+  - lines_per_sec: how many lines has the process written?
+"""
+import os
+import sys
+from datetime import datetime
+import time
+
+sidecar = os.environ.get("SIDECAR_FILE", "")
+total_lines = 0
+start_time = time.monotonic()
+
+while True:
+    ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
+    print((ts + "\n") * 1000)
+    total_lines += 1000
+    if sidecar:
+        elapsed = time.monotonic() - start_time
+        with open(sidecar, "w") as f:
+            f.write(f"{ts},{total_lines},{elapsed:.3f}\n")
+    time.sleep(0.01)
diff --git a/test/e2e/run_all_scenarios.sh b/test/e2e/run_all_scenarios.sh
new file mode 100644
index 000000000..689701b32
--- /dev/null
+++ b/test/e2e/run_all_scenarios.sh
@@ -0,0 +1,56 @@
+#!/bin/bash
+# Run all E2E flow control scenarios and generate HTML report.
+#
+# Usage: bash run_all_scenarios.sh [output_dir]
+#
+# Scenarios: trunk (~60s), backpressure (~60s), discard (~60s),
+#            discard-disconnect (~150s). Total: ~6 minutes.
+#
+# Output: <dir>/report.html, <dir>/*.jsonl, <dir>/*_meta.json
+
+set -e
+
+DIR="$(cd "$(dirname "$0")" && pwd)"
+REPO="$(cd "$DIR/../.." && pwd)"
+OUTDIR="${1:-$DIR/results}"
+LOCKFILE="/tmp/et_e2e_run.lock"
+
+exec 200>"$LOCKFILE"
+flock -n 200 || { echo "Another run in progress"; exit 1; }
+
+mkdir -p "$OUTDIR"
+echo "=== ET Flow Control E2E Test Suite ==="
+echo "Output: $OUTDIR"
+echo "Started: $(date)"
+echo ""
+
+kill -9 $(lsof -t -i:4444 2>/dev/null) $(lsof -t -i:4445 2>/dev/null) 2>/dev/null || true
+pkill -9 -f throttle_proxy 2>/dev/null || true
+sleep 2
+
+for SCENARIO in "trunk" "backpressure" "discard" "discard --disconnect"; do
+    MODE=$(echo "$SCENARIO" | awk '{print $1}')
+    EXTRA=$(echo "$SCENARIO" | awk '{$1=""; print $0}' | xargs)
+    echo "================================================================"
+    echo "  Scenario: $MODE $EXTRA"
+    echo "================================================================"
+
+    kill -9 $(lsof -t -i:4444 2>/dev/null) $(lsof -t -i:4445 2>/dev/null) 2>/dev/null || true
+    pkill -9 -f throttle_proxy 2>/dev/null || true
+    sleep 3
+
+    bash "$DIR/do_scenario.sh" $SCENARIO --outdir "$OUTDIR"
+    echo ""
+done
+
+cd "$REPO" && git checkout HEAD -- . 2>/dev/null
+cd build && cmake -DDISABLE_VCPKG=ON -GNinja .. 2>&1 | tail -1 && ninja -j4 2>&1 | tail -1
+
+echo "================================================================"
+echo "  Generating HTML report"
+echo "================================================================"
+python3 "$DIR/generate_report.py" "$OUTDIR" "$OUTDIR/report.html"
+
+echo ""
+echo "=== Complete at $(date) ==="
+ls -la "$OUTDIR"/*.html "$OUTDIR"/*.jsonl 2>/dev/null
diff --git a/test/e2e/run_e2e_test.sh b/test/e2e/run_e2e_test.sh
new file mode 100644
index 000000000..ea4d0a713
--- /dev/null
+++ b/test/e2e/run_e2e_test.sh
@@ -0,0 +1,102 @@
+#!/bin/bash
+# End-to-end flow control test. Runs etserver, a throttle proxy, etterminal,
+# and et client in tmux, then measures timestamp lag.
+#
+# Usage: bash run_e2e_test.sh [discard|backpressure] [rate_bytes_per_sec]
+#
+# Must be run outside of any process supervisor that kills children
+# (e.g. via cron). See test/e2e/README.md for details.
+
+MODE="${1:-discard}"
+RATE="${2:-100000}"
+DIR="$(cd "$(dirname "$0")" && pwd)"
+BUILD="$(cd "$DIR/../../build" && pwd)"
+SOCK="/tmp/et_e2e_$$.sock"
+FIFO="/tmp/et_e2e_$$.fifo"
+ID="E2E$(date +%s)"
+KEY="E2ETestKey123456789012345678901A"
+SIDECAR="/tmp/et_e2e_sidecar_$$.txt"
+
+cleanup() {
+    tmux -S "$SOCK" kill-server 2>/dev/null || true
+    kill -9 $(lsof -t -i:4444 2>/dev/null) 2>/dev/null || true
+    kill -9 $(lsof -t -i:4445 2>/dev/null) 2>/dev/null || true
+    pkill -9 -f "etterminal.*$ID" 2>/dev/null || true
+    rm -f "$FIFO" "$SOCK" "$SIDECAR" || true
+}
+trap cleanup EXIT
+cleanup 2>/dev/null
+
+# Start services
+tmux -S "$SOCK" new-session -d -s t -x 200 -y 50
+tmux -S "$SOCK" send-keys "$BUILD/etserver --serverfifo=$FIFO --port=4444" Enter
+sleep 5
+tmux -S "$SOCK" split-window -v
+sleep 1
+tmux -S "$SOCK" send-keys "python3 $DIR/throttle_proxy.py 4445 4444 $RATE" Enter
+sleep 3
+tmux -S "$SOCK" split-window -v
+sleep 1
+tmux -S "$SOCK" send-keys "$BUILD/etterminal --idpasskey='$ID/$KEY' --serverfifo=$FIFO" Enter
+sleep 5
+
+# Verify
+ES=$(lsof -i:4444 2>/dev/null | grep -c LISTEN)
+PX=$(lsof -i:4445 2>/dev/null | grep -c LISTEN)
+if [ "$ES" -lt 1 ] || [ "$PX" -lt 1 ]; then
+    echo "FAILED: etserver=$ES proxy=$PX"
+    exit 1
+fi
+
+# Connect client
+EXTRA=""
+if echo "$BUILD/et --help" | grep -q flow-control 2>/dev/null; then
+    EXTRA="--flow-control $MODE"
+fi
+tmux -S "$SOCK" new-window -n et
+tmux -S "$SOCK" send-keys -t t:et "$BUILD/et --idpasskey='$ID/$KEY' 127.0.0.1:4445 $EXTRA" Enter
+sleep 10
+
+# Verify proxy connection
+PROXY_OUT=$(tmux -S "$SOCK" capture-pane -t t:0.1 -p)
+if ! echo "$PROXY_OUT" | grep -q connect; then
+    echo "FAILED: proxy saw no connection"
+    echo "$PROXY_OUT" | tail -5
+    exit 1
+fi
+
+# Run timestamp test
+tmux -S "$SOCK" send-keys -t t:et "SIDECAR_FILE=$SIDECAR python3 $DIR/print_timestamps.py" Enter
+
+echo "=== $MODE mode, ${RATE}B/s proxy ==="
+for secs in 10 20 30; do
+    sleep 10
+    SCREEN=$(tmux -S "$SOCK" capture-pane -t t:et -p | grep -oE '[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]+' | tail -1)
+    REAL=$(python3 -c "from datetime import datetime; print(datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'))")
+    if [ -n "$SCREEN" ]; then
+        LAG=$(python3 -c "
+from datetime import datetime
+s=datetime.strptime('${SCREEN}'.strip(),'%Y-%m-%d %H:%M:%S.%f')
+r=datetime.strptime('${REAL}'.strip(),'%Y-%m-%d %H:%M:%S.%f')
+print(f'{(r-s).total_seconds():.1f}')")
+        echo "  t=${secs}s: display_lag=${LAG}s"
+    else
+        echo "  t=${secs}s: no data on screen"
+    fi
+
+    # Check if the process is stalled by reading the sidecar
+    if [ -f "$SIDECAR" ]; then
+        PROCESS_TS=$(cat "$SIDECAR")
+        PROCESS_LAG=$(python3 -c "
+from datetime import datetime
+s=datetime.strptime('${PROCESS_TS}'.strip(),'%Y-%m-%d %H:%M:%S.%f')
+r=datetime.strptime('${REAL}'.strip(),'%Y-%m-%d %H:%M:%S.%f')
+print(f'{(r-s).total_seconds():.1f}')")
+        echo "           process_lag=${PROCESS_LAG}s (0=running, >1=stalled)"
+    fi
+done
+
+echo ""
+echo "Proxy stats:"
+tmux -S "$SOCK" capture-pane -t t:0.1 -p | grep -E 'sent=|connect' | tail -3
+echo "etserver alive: $(lsof -i:4444 2>/dev/null | grep -c LISTEN)"
diff --git a/test/e2e/throttle_proxy.py b/test/e2e/throttle_proxy.py
new file mode 100644
index 000000000..de5da5e66
--- /dev/null
+++ b/test/e2e/throttle_proxy.py
@@ -0,0 +1,159 @@
+"""TCP throttle proxy for E2E testing. Applies TCP backpressure to the server.
+
+Reads from the server at a limited rate, causing the kernel TCP send buffer
+to fill up. Client->server direction (keystrokes) forwarded at full speed.
+
+Supports disconnect simulation:
+  SIGUSR1 -> kill all connections, block new ones for 60s
+  SIGUSR2 -> resume immediately
+
+Usage: python3 throttle_proxy.py <listen_port> <target_port> [bytes_per_sec]
+"""
+import socket
+import sys
+import threading
+import time
+import signal
+import select
+from datetime import datetime
+
+LISTEN_PORT, TARGET_PORT = int(sys.argv[1]), int(sys.argv[2])
+RATE = int(sys.argv[3]) if len(sys.argv) > 3 else 100000
+
+disconnect_until = 0
+active_connections = []
+lock = threading.Lock()
+
+
+def log(msg):
+    ts = datetime.now().strftime("%H:%M:%S.%f")
+    print(f"[proxy {ts}] {msg}", flush=True)
+
+
+def handle_disconnect(signum, frame):
+    global disconnect_until
+    disconnect_until = time.monotonic() + 60
+    log("DISCONNECT: killing all connections for 60s")
+    with lock:
+        for cli, srv, stop in active_connections:
+            stop.set()
+            try:
+                cli.close()
+            except Exception:
+                pass
+            try:
+                srv.close()
+            except Exception:
+                pass
+        active_connections.clear()
+
+
+def handle_reconnect(signum, frame):
+    global disconnect_until
+    disconnect_until = 0
+    log("RECONNECT: accepting connections again")
+
+
+signal.signal(signal.SIGUSR1, handle_disconnect)
+signal.signal(signal.SIGUSR2, handle_reconnect)
+
+
+def throttled_forward(name, src, dst, rate, stop):
+    total = 0
+    CHUNK = max(rate // 20, 64)
+    last_t = time.time()
+    try:
+        while not stop.is_set():
+            data = src.recv(CHUNK)
+            if not data:
+                break
+            dst.sendall(data)
+            total += len(data)
+            time.sleep(len(data) / rate)
+            now = time.time()
+            if now - last_t > 2:
+                log(f"{name}: sent={total:,}B")
+                last_t = now
+    except Exception:
+        pass
+    stop.set()
+
+
+def fast_forward(name, src, dst, stop):
+    try:
+        while not stop.is_set():
+            data = src.recv(4096)
+            if not data:
+                break
+            dst.sendall(data)
+    except Exception:
+        pass
+    stop.set()
+
+
+def handle(cli, addr):
+    log(f"connect {addr}")
+    srv = None
+    stop = threading.Event()
+    try:
+        srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        # Clamp the receive buffer BEFORE connect (window scaling is
+        # negotiated at handshake). Without this, the proxy's kernel rcvbuf
+        # autotunes to multiple MB and silently absorbs the server's output,
+        # hiding buffer bloat that a real slow link would push back on.
+        # 64KB ~= 0.64s of queue at the default 100KB/s rate.
+        srv.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 64 * 1024)
+        srv.connect(("127.0.0.1", TARGET_PORT))
+        log(f"srv rcvbuf={srv.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)}")
+        with lock:
+            active_connections.append((cli, srv, stop))
+        t1 = threading.Thread(
+            target=throttled_forward,
+            args=("s->c", srv, cli, RATE, stop),
+            daemon=True,
+        )
+        t2 = threading.Thread(
+            target=fast_forward, args=("c->s", cli, srv, stop), daemon=True
+        )
+        t1.start()
+        t2.start()
+        t1.join()
+        t2.join()
+    except Exception as e:
+        log(f"error: {e}")
+    finally:
+        for s in (cli, srv):
+            try:
+                s.close()
+            except Exception:
+                pass
+        with lock:
+            active_connections[:] = [
+                (c, s, st) for c, s, st in active_connections if st is not stop
+            ]
+    log(f"disconnect {addr}")
+
+
+# Dual-stack listener
+s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
+s.bind(("::", LISTEN_PORT))
+s.listen(5)
+s.setblocking(False)
+log(f"[::]:{LISTEN_PORT} -> 127.0.0.1:{TARGET_PORT} rate={RATE}B/s")
+
+while True:
+    try:
+        r, _, _ = select.select([s], [], [], 1.0)
+        for _ in r:
+            c, a = s.accept()
+            if time.monotonic() < disconnect_until:
+                log(f"rejecting {a} (disconnected)")
+                c.close()
+                continue
+            threading.Thread(target=handle, args=(c, a), daemon=True).start()
+    except KeyboardInterrupt:
+        break
+    except Exception:
+        pass
diff --git a/test/e2e/throughput_test.sh b/test/e2e/throughput_test.sh
new file mode 100644
index 000000000..c3d8439a8
--- /dev/null
+++ b/test/e2e/throughput_test.sh
@@ -0,0 +1,72 @@
+#!/bin/bash
+# Measures bulk throughput through the full et pipeline on a fast link
+# (no throttle proxy). Used to verify that flow control and kernel buffer
+# tuning (TCP_NOTSENT_LOWAT, SO_SNDBUF clamps) do not regress throughput.
+#
+# Usage: bash throughput_test.sh [--flow-control <mode>] [--bytes N]
+# Assumes the build/ directory contains current binaries.
+
+FC_FLAG=""
+BYTES=30000000
+while [ $# -gt 0 ]; do
+    case "$1" in
+        --flow-control) FC_FLAG="--flow-control $2"; shift ;;
+        --bytes) BYTES="$2"; shift ;;
+    esac
+    shift
+done
+
+DIR="$(cd "$(dirname "$0")" && pwd)"
+REPO="$(cd "$DIR/../.." && pwd)"
+BUILD="$REPO/build"
+PORT=4446
+ID="TPT$(date +%s | tail -c 8)"
+KEY="E2ETestKey123456789012345678901A"
+ETMUX="/tmp/et_tpt_client_$$.sock"
+FIFO="/tmp/et_tpt_$$.fifo"
+
+cleanup() {
+    tmux -S "$ETMUX" kill-server 2>/dev/null
+    kill $ETSERVER_PID 2>/dev/null
+    kill -9 $(lsof -t -i:$PORT 2>/dev/null) 2>/dev/null
+    pkill -9 -f "etterminal.*$ID" 2>/dev/null
+    rm -f "$FIFO" "$ETMUX"
+}
+trap cleanup EXIT
+
+"$BUILD/etserver" --serverfifo="$FIFO" --port=$PORT --logdir=/tmp &>/dev/null &
+ETSERVER_PID=$!
+sleep 2
+"$BUILD/etterminal" --idpasskey="$ID/$KEY" --serverfifo="$FIFO" --logdir=/tmp &>/dev/null &
+sleep 2
+
+tmux -S "$ETMUX" new-session -d -s et -x 200 -y 50
+for i in $(seq 1 30); do
+    tmux -S "$ETMUX" capture-pane -p 2>/dev/null | grep -q '\$' && break
+    sleep 1
+done
+tmux -S "$ETMUX" send-keys "$BUILD/et --idpasskey='$ID/$KEY' 127.0.0.1:$PORT $FC_FLAG" Enter
+for i in $(seq 1 30); do
+    tmux -S "$ETMUX" capture-pane -p 2>/dev/null | grep -qF '~]$' && break
+    sleep 1
+done
+
+# The marker is split in the typed command so the echoed command line
+# itself doesn't match the completion grep.
+T0=$(python3 -c 'import time; print(f"{time.time():.3f}")')
+tmux -S "$ETMUX" send-keys "base64 -w 200 /dev/zero | head -c $BYTES; echo 'TPT_''END_MARK'" Enter
+ELAPSED="timeout"
+for i in $(seq 1 1200); do
+    sleep 0.1
+    if tmux -S "$ETMUX" capture-pane -p 2>/dev/null | grep -qF 'TPT_END_MARK'; then
+        ELAPSED=$(python3 -c "import time; print(f'{time.time() - $T0:.2f}')")
+        break
+    fi
+done
+
+if [ "$ELAPSED" = "timeout" ]; then
+    echo "RESULT: timeout"
+else
+    MBPS=$(python3 -c "print(f'{$BYTES / float('$ELAPSED') / 1e6:.1f}')")
+    echo "RESULT: ${BYTES} bytes in ${ELAPSED}s = ${MBPS} MB/s (flow-control: ${FC_FLAG:-default})"
+fi
diff --git a/test/integration_tests/TerminalServerTest.cpp b/test/integration_tests/TerminalServerTest.cpp
index 5adab3a2c..63dcdd5b9 100644
--- a/test/integration_tests/TerminalServerTest.cpp
+++ b/test/integration_tests/TerminalServerTest.cpp
@@ -1,4 +1,5 @@
 #include <atomic>
+#include <catch2/generators/catch_generators.hpp>
 
 #include "FakeConsole.hpp"
 #include "FakeSshSetupHandler.hpp"
@@ -479,6 +480,53 @@ TEST_CASE_METHOD(ServerEndToEndTestFixture, "ServerDataTransferTest",
   sshSetupHandler->shutdownHandler();
 }
 
+// Sessions that opt into flow control (backpressure or discard) must
+// still deliver data intact in both directions. The volumes here stay far
+// below WriteBuffer::MAX_BUFFER_SIZE, so even discard mode must not drop
+// anything.
+TEST_CASE_METHOD(ServerEndToEndTestFixture, "ServerFlowControlDataTransferTest",
+                 "[ServerFlowControlDataTransferTest][integration]") {
+  const et::FlowControlMode mode =
+      GENERATE(et::FLOW_CONTROL_BACKPRESSURE, et::FLOW_CONTROL_DISCARD);
+  CAPTURE(mode);
+
+  auto [id, passkey] = sshSetupHandler->SetupSsh(
+      "", "localhost", "localhost", 2022, "", "", false, 0, "", "", {});
+
+  sleep(1);
+
+  shared_ptr<TerminalClient> terminalClient(new TerminalClient(
+      clientSocketHandler, clientPipeSocketHandler, serverEndpoint, id, passkey,
+      fakeConsole, false, "", "", false, "", MAX_CLIENT_KEEP_ALIVE_DURATION, {},
+      mode));
+  thread terminalClientThread(
+      [terminalClient]() { terminalClient->run("", false); });
+  sleep(3);
+
+  // Keyboard input path (console -> terminal)
+  string s = "flow_control_input";
+  for (size_t a = 0; a < s.size(); a++) {
+    fakeConsole->simulateKeystrokes(string(1, s[a]));
+  }
+  string resultConcat;
+  for (size_t a = 0; a < s.size(); a++) {
+    resultConcat = resultConcat.append(fakeUserTerminal->getKeystrokes(1));
+  }
+  REQUIRE(resultConcat == s);
+
+  // Terminal output path (terminal -> console); this is the direction the
+  // flow-control buffering applies to.
+  string out = "flow_control_output";
+  fakeUserTerminal->simulateTerminalResponse(out);
+  REQUIRE(fakeConsole->getTerminalData(out.length()) == out);
+
+  terminalClient->shutdown();
+  terminalClientThread.join();
+  terminalClient.reset();
+
+  sshSetupHandler->shutdownHandler();
+}
+
 TEST_CASE_METHOD(ServerEndToEndTestFixture, "ServerJumphostTest",
                  "[ServerJumphostTest][integration]") {
   // The destination terminal is already set up in the fixture at
diff --git a/test/unit_tests/WriteBufferTest.cpp b/test/unit_tests/WriteBufferTest.cpp
new file mode 100644
index 000000000..5ee6fa266
--- /dev/null
+++ b/test/unit_tests/WriteBufferTest.cpp
@@ -0,0 +1,181 @@
+#include "TestHeaders.hpp"
+#include "WriteBuffer.hpp"
+
+using namespace et;
+
+TEST_CASE("WriteBuffer basic operations", "[WriteBuffer]") {
+  WriteBuffer buffer;
+
+  SECTION("Empty buffer state") {
+    REQUIRE(buffer.canAcceptMore() == true);
+    REQUIRE(buffer.hasPendingData() == false);
+    REQUIRE(buffer.size() == 0);
+
+    size_t count;
+    const char* data = buffer.peekData(&count);
+    REQUIRE(data == nullptr);
+    REQUIRE(count == 0);
+  }
+
+  SECTION("Enqueue and peek") {
+    buffer.enqueue("hello");
+    REQUIRE(buffer.hasPendingData() == true);
+    REQUIRE(buffer.size() == 5);
+
+    size_t count;
+    const char* data = buffer.peekData(&count);
+    REQUIRE(data != nullptr);
+    REQUIRE(count == 5);
+    REQUIRE(string(data, count) == "hello");
+  }
+
+  SECTION("Consume partial") {
+    buffer.enqueue("hello");
+    buffer.consume(2);
+    REQUIRE(buffer.size() == 3);
+
+    size_t count;
+    const char* data = buffer.peekData(&count);
+    REQUIRE(count == 3);
+    REQUIRE(string(data, count) == "llo");
+  }
+
+  SECTION("Consume full chunk") {
+    buffer.enqueue("hello");
+    buffer.enqueue("world");
+    REQUIRE(buffer.size() == 10);
+
+    buffer.consume(5);
+    REQUIRE(buffer.size() == 5);
+
+    size_t count;
+    const char* data = buffer.peekData(&count);
+    REQUIRE(count == 5);
+    REQUIRE(string(data, count) == "world");
+  }
+
+  SECTION("Consume across chunks") {
+    buffer.enqueue("abc");
+    buffer.enqueue("defgh");
+    REQUIRE(buffer.size() == 8);
+
+    buffer.consume(5);  // Consumes "abc" + "de"
+    REQUIRE(buffer.size() == 3);
+
+    size_t count;
+    const char* data = buffer.peekData(&count);
+    REQUIRE(count == 3);
+    REQUIRE(string(data, count) == "fgh");
+  }
+
+  SECTION("Clear buffer") {
+    buffer.enqueue("hello");
+    buffer.enqueue("world");
+    buffer.clear();
+
+    REQUIRE(buffer.hasPendingData() == false);
+    REQUIRE(buffer.size() == 0);
+    REQUIRE(buffer.canAcceptMore() == true);
+  }
+
+  SECTION("Empty string enqueue is ignored") {
+    buffer.enqueue("");
+    REQUIRE(buffer.hasPendingData() == false);
+    REQUIRE(buffer.size() == 0);
+  }
+}
+
+TEST_CASE("WriteBuffer backpressure", "[WriteBuffer]") {
+  WriteBuffer buffer(WriteBufferMode::BACKPRESSURE);
+
+  SECTION("canAcceptMore returns false when buffer is full") {
+    // Fill the buffer to capacity
+    string largeChunk(WriteBuffer::MAX_BUFFER_SIZE, 'x');
+    buffer.enqueue(largeChunk);
+
+    REQUIRE(buffer.canAcceptMore() == false);
+    REQUIRE(buffer.size() == WriteBuffer::MAX_BUFFER_SIZE);
+
+    // Consume some data
+    buffer.consume(1024);
+    REQUIRE(buffer.canAcceptMore() == true);
+  }
+
+  SECTION("default mode is backpressure") {
+    WriteBuffer defaultBuffer;
+    REQUIRE(defaultBuffer.getMode() == WriteBufferMode::BACKPRESSURE);
+  }
+}
+
+TEST_CASE("WriteBuffer discard mode", "[WriteBuffer]") {
+  WriteBuffer buffer(WriteBufferMode::DISCARD);
+
+  SECTION("canAcceptMore always returns true") {
+    string largeChunk(WriteBuffer::MAX_BUFFER_SIZE, 'x');
+    buffer.enqueue(largeChunk);
+
+    // In discard mode, canAcceptMore is always true
+    REQUIRE(buffer.canAcceptMore() == true);
+  }
+
+  SECTION("old data is discarded when buffer overflows") {
+    // Fill with chunk A (128KB)
+    const size_t halfSize = WriteBuffer::MAX_BUFFER_SIZE / 2;
+    string chunkA(halfSize, 'A');
+    buffer.enqueue(chunkA);
+    REQUIRE(buffer.size() == halfSize);
+
+    // Fill with chunk B (128KB) -- total now equals MAX
+    string chunkB(halfSize, 'B');
+    buffer.enqueue(chunkB);
+    REQUIRE(buffer.size() == WriteBuffer::MAX_BUFFER_SIZE);
+
+    // Add chunk C (128KB) -- should discard chunk A
+    string chunkC(halfSize, 'C');
+    buffer.enqueue(chunkC);
+
+    // Buffer should be within MAX_BUFFER_SIZE
+    REQUIRE(buffer.size() <= WriteBuffer::MAX_BUFFER_SIZE);
+
+    // The oldest data (chunk A) should be gone
+    // The front of the buffer should now be chunk B
+    size_t count;
+    const char* data = buffer.peekData(&count);
+    REQUIRE(data != nullptr);
+    REQUIRE(data[0] == 'B');
+  }
+
+  SECTION("buffer stays bounded with many enqueues") {
+    const size_t chunkSize = 4096;
+    // Enqueue much more than MAX_BUFFER_SIZE
+    for (size_t i = 0; i < WriteBuffer::MAX_BUFFER_SIZE * 4; i += chunkSize) {
+      string chunk(chunkSize, 'A' + ((i / chunkSize) % 26));
+      buffer.enqueue(chunk);
+      // Buffer should never exceed MAX_BUFFER_SIZE + one chunk
+      REQUIRE(buffer.size() <= WriteBuffer::MAX_BUFFER_SIZE + chunkSize);
+    }
+  }
+
+  SECTION("newest data is preserved") {
+    // Fill buffer well past capacity
+    const size_t chunkSize = 1024;
+    string lastChunk;
+    for (size_t i = 0; i < WriteBuffer::MAX_BUFFER_SIZE * 2; i += chunkSize) {
+      lastChunk = string(chunkSize, 'A' + ((i / chunkSize) % 26));
+      buffer.enqueue(lastChunk);
+    }
+
+    // Drain the buffer and verify the last chunk is in the output
+    string allData;
+    while (buffer.hasPendingData()) {
+      size_t count;
+      const char* data = buffer.peekData(&count);
+      allData.append(data, count);
+      buffer.consume(count);
+    }
+
+    // The tail of the drained data should be our last chunk
+    REQUIRE(allData.size() >= lastChunk.size());
+    REQUIRE(allData.substr(allData.size() - lastChunk.size()) == lastChunk);
+  }
+}