diff --git a/WORKSPACE b/WORKSPACE
index 6efad5d57..153ac6c5e 100644
--- a/WORKSPACE
+++ b/WORKSPACE
@@ -45,6 +45,8 @@ git_repository(
         "@//patches:0006-test-integration-Defer-fake-upstream-read-enable-un.patch",
         "@//patches:0007-config-add-grpc-mux-stream-event-callback.patch",
         "@//patches:0008-repo-Make-yq-dependency-optional-for-CI-config-parsi.patch",
+        "@//patches:0009-network-Allow-write-filters-to-consume-data-after-e.patch",
+        "@//patches:0010-tcp_proxy-Allow-filters-to-wait-for-upstream-close.patch",
     ],
     # // clang-format off: Envoy's format check: Only repository_locations.bzl may contains URL references
     remote = "https://github.com/envoyproxy/envoy.git",
diff --git a/cilium/BUILD b/cilium/BUILD
index cc2fac167..ef1888b16 100644
--- a/cilium/BUILD
+++ b/cilium/BUILD
@@ -160,6 +160,7 @@ envoy_cc_library(
         "//cilium:accesslog_lib",
         "//cilium:filter_state_lib",
         "//cilium/api:websocket_cc_proto",
+        "@com_google_protobuf//third_party/utf8_range:utf8_validity",
         "@envoy//bazel/external/http_parser",
         "@envoy//envoy/common/crypto:crypto_interface",
         "@envoy//source/common/common:base64_lib",
@@ -171,6 +172,7 @@ envoy_cc_library(
         "@envoy//source/common/http:utility_lib",
         "@envoy//source/common/network:filter_manager_lib",
         "@envoy//source/common/stream_info:bool_accessor_lib",
+        "@envoy//source/common/stream_info:uint64_accessor_lib",
         "@envoy//source/common/tcp_proxy",
         "@envoy_api//envoy/extensions/request_id/uuid/v3:pkg_cc_proto",
     ],
diff --git a/cilium/websocket.cc b/cilium/websocket.cc
index c84a2b279..47ce3b233 100644
--- a/cilium/websocket.cc
+++ b/cilium/websocket.cc
@@ -1,5 +1,6 @@
 #include "cilium/websocket.h"
 
+#include <chrono>
 #include <cstdint>
 #include <memory>
 #include <string>
@@ -19,6 +20,7 @@
 #include "source/common/protobuf/protobuf.h"
 #include "source/common/protobuf/utility.h"
 #include "source/common/stream_info/bool_accessor_impl.h"
+#include "source/common/stream_info/uint64_accessor_impl.h"
 #include "source/common/tcp_proxy/tcp_proxy.h"
 
 #include "absl/status/statusor.h"
@@ -34,6 +36,8 @@ namespace WebSocket {
 
 namespace {
 
+constexpr std::chrono::milliseconds WebSocketTransportCloseTimeout{1000};
+
 Http::RegisterCustomInlineHeader<Http::CustomInlineHeaderRegistry::Type::RequestHeaders>
     origin_handle(Http::CustomHeaders::get().Origin);
 Http::RegisterCustomInlineHeader<Http::CustomInlineHeaderRegistry::Type::RequestHeaders>
@@ -128,6 +132,16 @@ void Instance::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callb
   callbacks_->connection().streamInfo().filterState()->setData(
       TcpProxy::ReceiveBeforeConnectKey, std::make_unique<StreamInfo::BoolAccessorImpl>(true),
       StreamInfo::FilterState::StateType::Mutable, StreamInfo::FilterState::LifeSpan::Connection);
+
+  // After both directions of a client-side WebSocket tunnel have ended, TcpProxy must flush the
+  // final data and CLOSE frame and wait for the peer's transport FIN. Closing immediately can
+  // generate an RST if a peer control frame is still unread, losing the just-flushed frames.
+  if (config_->client_) {
+    callbacks_->connection().streamInfo().filterState()->setData(
+        TcpProxy::UpstreamFlushWaitTimeoutMs,
+        std::make_unique<StreamInfo::UInt64AccessorImpl>(WebSocketTransportCloseTimeout.count()),
+        StreamInfo::FilterState::StateType::Mutable, StreamInfo::FilterState::LifeSpan::Connection);
+  }
 }
 
 Network::FilterStatus Instance::onNewConnection() {
diff --git a/cilium/websocket_codec.cc b/cilium/websocket_codec.cc
index b7106b2e2..406b7d220 100644
--- a/cilium/websocket_codec.cc
+++ b/cilium/websocket_codec.cc
@@ -37,6 +37,7 @@
 #include "absl/strings/string_view.h"
 #include "cilium/websocket_config.h"
 #include "cilium/websocket_protocol.h"
+#include "third_party/utf8_range/utf8_validity.h"
 
 namespace Envoy {
 namespace Cilium {
@@ -373,15 +374,45 @@ void Codec::handshake() {
 void Codec::encode(Buffer::Instance& data, bool end_stream) {
   ENVOY_LOG(debug, "websocket: encode {} bytes, end_stream: {}", data.length(), end_stream);
 
-  encoder_.encode(data, end_stream, OPCODE_BIN);
+  // RFC 6455 section 5.5.1 forbids sending data frames after a CLOSE frame.
+  if (close_sent_) {
+    const auto len = data.length();
+    if (len > 0) {
+      ENVOY_LOG(debug, "websocket encoder: data received after CLOSE: {} bytes", len);
+      data.drain(len);
+    }
+    return;
+  }
+
+  encoder_.encode(data, OPCODE_BIN);
+
+  if (end_stream) {
+    // CLOSE represents the FIN for this direction of the tunneled TCP connection. If the peer
+    // half-closed first, echo its validated CLOSE payload only after all reverse-direction data
+    // has been encoded. This deliberately delays the CLOSE response to preserve TCP half-close
+    // semantics when the WebSocket infrastructure permits it.
+    Buffer::OwnedImpl close_payload;
+    close_payload.add(decoder_.close_payload_);
+    encoder_.encode(close_payload, OPCODE_CLOSE);
+    close_sent_ = true;
+
+    // Only end stream if CLOSE has been received already and if we are the server, as additional
+    // control frames may still be sent, and RFC 6455 section 7.1.1 recommends that the server close
+    // the underlying transport after both sides have sent CLOSE. Keep the client transport open so
+    // that its final frames are flushed before the server closes the connection.
+    end_stream = !config()->client_ && decoder_.close_received_;
+  }
 
   // Only forward data if handshake has completed
   if (accepted_) {
-    // Reset idle timer on data
+    // Reset idle timer on any frame content
     if (encoder_.hasData()) {
       resetPingTimer();
     }
-    parent_->injectEncoded(encoder_.data(), encoder_.endStream());
+    if (end_stream) {
+      encoded_end_stream_sent_ = end_stream;
+    }
+    parent_->injectEncoded(encoder_.data(), end_stream);
   }
 }
 
@@ -586,9 +617,9 @@ void Codec::decode(Buffer::Instance& data, bool end_stream) {
         return closeOnError(handshake_buffer_, "Invalid WebSocket response");
       }
 
-      // Kick write on the other direction
-      parent_->injectEncoded(encoder_.data(), encoder_.endStream());
-
+      // Kick the buffered write. This is the client codec, so the server remains responsible for
+      // ending the underlying WebSocket transport.
+      parent_->injectEncoded(encoder_.data(), false);
     } else {
       // Server needs to wait for a valid handshake request before accepting any data
       RequestParser parser;
@@ -654,118 +685,120 @@ void Codec::decode(Buffer::Instance& data, bool end_stream) {
   // Handshake done, process data.
   decoder_.decode(data, end_stream);
 
+  if (decoder_.protocol_error_) {
+    config->stats_.protocol_error_.inc();
+    decoder_.drain();
+    return closeOnError("invalid WebSocket control frame");
+  }
+
+  if (end_stream && !decoder_.close_received_) {
+    decoder_.drain();
+    return closeOnError("websocket transport closed without CLOSE");
+  }
+
   // Reset idle timer on data
   if (decoder_.hasData()) {
     resetPingTimer();
   }
 
-  parent_->injectDecoded(decoder_.data(), decoder_.endStream());
+  // The decoder records CLOSE before injecting the corresponding TCP FIN. That injection may
+  // synchronously report the reverse-direction FIN through encode(..., end_stream=true).
+  const bool decoded_end_stream = decoder_.close_received_ && !decoded_end_stream_sent_;
+  parent_->injectDecoded(decoder_.data(), decoded_end_stream);
+  decoded_end_stream_sent_ |= decoded_end_stream;
+
+  // Complete CLOSE handshake by encoding end_stream from the server side, if closed in both
+  // directions. RFC 6455 section 7.1.1 recommends that the client wait for the server to close the
+  // connection.
+  if (!config->client_ && decoder_.close_received_ && close_sent_ && !encoded_end_stream_sent_) {
+    Buffer::OwnedImpl empty;
+    encoded_end_stream_sent_ = true;
+    parent_->injectEncoded(empty, true);
+  }
+  // Otherwise defer the CLOSE response until encode(..., end_stream=true) reports the FIN from
+  // the other direction of the tunneled TCP connection. Data remains legal to encode until that
+  // response is sent. RFC 6455 section 5.5.1 permits delaying a CLOSE response, but warns that a
+  // generic peer is not guaranteed to keep processing data after sending CLOSE; this behavior is
+  // therefore a tunnel convention between cooperating endpoints.
 }
 
 bool Codec::ping(const void* payload, size_t len) {
-  if (encoder_.endStream()) {
+  // RFC 6455 section 5.5.2 permits PING until the connection is closed. For this tunnel, one CLOSE
+  // is only a directional FIN, so keepalive PINGs continue through the half-closed interval.
+  if (close_sent_ && decoder_.close_received_) {
     return false;
   }
   Buffer::OwnedImpl buf(payload, len);
-  encoder_.encode(buf, false, OPCODE_PING);
+  encoder_.encode(buf, OPCODE_PING);
   parent_->config()->stats_.ping_sent_count_.inc();
-  parent_->injectEncoded(encoder_.data(), encoder_.endStream());
+  parent_->injectEncoded(encoder_.data(), false);
   return true;
 }
 
 bool Codec::pong(const void* payload, size_t len) {
-  if (encoder_.endStream()) {
+  // RFC 6455 section 5.5.2 does not require PONG after receiving CLOSE, but permits it. Continue to
+  // answer PING while the tunneled TCP connection is only half-closed.
+  if (close_sent_ && decoder_.close_received_) {
     return false;
   }
   Buffer::OwnedImpl buf(payload, len);
-  encoder_.encode(buf, false, OPCODE_PONG);
-  parent_->injectEncoded(encoder_.data(), encoder_.endStream());
+  encoder_.encode(buf, OPCODE_PONG);
+  parent_->injectEncoded(encoder_.data(), false);
   return true;
 }
 
 // Encoder
 
-// Encode 'data' and 'end_stream' as websocket frames into 'encoded_'. Uses 'opcode' as the
-// websocket frame type for the data frames.
-void Codec::Encoder::encode(Buffer::Instance& data, bool end_stream, uint8_t opcode) {
+// Encode 'data' as one websocket frame into 'encoded_'.
+void Codec::Encoder::encode(Buffer::Instance& data, uint8_t opcode) {
   auto hex_len = std::min(data.length(), 20UL);
   const uint8_t* hex_data = reinterpret_cast<uint8_t*>(data.linearize(hex_len));
-  ENVOY_LOG(debug, "websocket encoder: {} bytes: 0x{}, end_stream: {}, opcode: {}", data.length(),
-            Hex::encode(hex_data, hex_len), end_stream, opcode);
+  ENVOY_LOG(debug, "websocket encoder: {} bytes: 0x{}, opcode: {}", data.length(),
+            Hex::encode(hex_data, hex_len), opcode);
 
   auto& config = parent_.config();
-  //
-  // Encode data as a single WebSocket frame
-  //
-  if (data.length() > 0) {
-    uint8_t frame_header[14];
-    size_t frame_header_length = 2;
-    size_t payload_len = data.length();
-
-    frame_header[0] = FIN_MASK | opcode;
-    if (payload_len < 126) {
-      frame_header[1] = payload_len;
-    } else if (payload_len < 65536) {
-      uint16_t len16;
-
-      frame_header[1] = 126;
-      len16 = htobe16(payload_len);
-      memcpy(frame_header + frame_header_length, &len16, 2); // NOLINT(safe-memcpy)
-      frame_header_length += 2;
-    } else {
-      uint64_t len64;
-
-      frame_header[1] = 127;
-      len64 = htobe64(payload_len);
-      memcpy(frame_header + frame_header_length, &len64, 8); // NOLINT(safe-memcpy)
-      frame_header_length += 8;
-    }
-
-    // Client must mask the payload
-    if (config->client_) {
-      frame_header[1] |= MASK_MASK;
-
-      union {
-        uint8_t bytes[4];
-        uint32_t word;
-      } mask;
-
-      mask.word = config->random_.random();
-      memcpy(frame_header + frame_header_length, &mask, 4); // NOLINT(safe-memcpy)
-      frame_header_length += 4;
-      uint8_t* buf = reinterpret_cast<uint8_t*>(data.linearize(payload_len));
-      maskData(buf, payload_len, mask.bytes);
-    }
-
-    // Add frame header and (masked) data
-    encoded_.add(absl::string_view{reinterpret_cast<char*>(frame_header), frame_header_length});
-    encoded_.move(data, payload_len);
+  const size_t payload_len = data.length();
+  const bool control_frame = opcode >= OPCODE_CLOSE;
+  if (payload_len == 0 && !control_frame) {
+    return;
   }
 
-  //
-  // Append closing frame if 'end_stream'
-  //
-  if (end_stream) {
-    uint8_t frame_header[14];
-    size_t frame_header_length = 2;
-    size_t payload_len = 0;
+  RELEASE_ASSERT(!control_frame || payload_len <= WEBSOCKET_CONTROL_FRAME_MAX_SIZE,
+                 "websocket control frame too large");
 
-    frame_header[0] = FIN_MASK | OPCODE_CLOSE;
+  uint8_t frame_header[14];
+  size_t frame_header_length = 2;
+  frame_header[0] = FIN_MASK | opcode;
+  if (payload_len < 126) {
     frame_header[1] = payload_len;
-    // Client must mask the payload
-    if (config->client_) {
-      frame_header[1] |= MASK_MASK;
-
-      uint32_t mask = config->random_.random();
-      memcpy(frame_header + frame_header_length, &mask, 4); // NOLINT(safe-memcpy)
-      frame_header_length += 4;
-      // No data to mask
+  } else if (payload_len < 65536) {
+    frame_header[1] = 126;
+    frame_header[frame_header_length++] = static_cast<uint8_t>(payload_len >> 8);
+    frame_header[frame_header_length++] = static_cast<uint8_t>(payload_len);
+  } else {
+    frame_header[1] = 127;
+    const uint64_t encoded_payload_len = payload_len;
+    for (int shift = 56; shift >= 0; shift -= 8) {
+      frame_header[frame_header_length++] = static_cast<uint8_t>(encoded_payload_len >> shift);
     }
-    encoded_.add(reinterpret_cast<void*>(frame_header), frame_header_length);
-    end_stream_ = true;
+  }
 
-    ENVOY_LOG(debug, "websocket encoder: sent WebSocket CLOSE message, end_stream: {}", end_stream);
+  // WebSocket clients must mask all frames, including control frames.
+  if (config->client_) {
+    frame_header[1] |= MASK_MASK;
+    const uint32_t mask_word = config->random_.random();
+    uint8_t mask[4];
+    for (size_t i = 0; i < sizeof(mask); ++i) {
+      mask[i] = static_cast<uint8_t>(mask_word >> (i * 8));
+      frame_header[frame_header_length++] = mask[i];
+    }
+    uint8_t* buf = reinterpret_cast<uint8_t*>(data.linearize(payload_len));
+    maskData(buf, payload_len, mask);
   }
+
+  // Add frame header and (maybe masked) data
+  encoded_.add(absl::string_view{reinterpret_cast<char*>(frame_header), frame_header_length});
+  encoded_.move(data, payload_len);
 }
 
 // Decoder
@@ -792,40 +825,39 @@ void Codec::Decoder::decode(Buffer::Instance& data, bool end_stream) {
 
   buffer_.move(data);
 
-  if (end_stream_ && buffer_.length() > 0) {
-    ENVOY_LOG(debug, "websocket decoder: data received after CLOSE: {} bytes", buffer_.length());
-    buffer_.drain(buffer_.length());
-    return;
-  }
-
-  if (end_stream) {
-    end_stream_ = true;
-  }
-
   while (buffer_.length() > 0) {
-    // Try finish any frame in progress
+    // Try finish any data frame in progress
     while (payload_remaining_ > 0) {
       auto slice = buffer_.frontSlice();
       size_t n_bytes = std::min(slice.len_, payload_remaining_);
 
-      // Unmask data in place
-      uint8_t* buf = static_cast<uint8_t*>(slice.mem_);
-      auto hex_len = std::min(n_bytes, 20UL);
-      if (unmasking_) {
-        ENVOY_LOG(
-            trace,
-            "websocket decoder: unmasking payload remaining: {}, offset: {}, processing: {}: 0x{}",
-            payload_remaining_, payload_offset_, n_bytes, Hex::encode(buf, hex_len));
-        payload_offset_ = maskData(buf, n_bytes, mask_, payload_offset_);
+      if (n_bytes > 0) {
+        // Unmask data in place
+        uint8_t* buf = static_cast<uint8_t*>(slice.mem_);
+        auto hex_len = std::min(n_bytes, 20UL);
+        if (unmasking_) {
+          ENVOY_LOG(trace,
+                    "websocket decoder: unmasking payload remaining: {}, offset: {}, processing: "
+                    "{}: 0x{}",
+                    payload_remaining_, payload_offset_, n_bytes, Hex::encode(buf, hex_len));
+          payload_offset_ = maskData(buf, n_bytes, mask_, payload_offset_);
+        }
+        ENVOY_LOG(trace,
+                  "websocket decoder: payload remaining: {}, offset: {}, processing: {}: 0x{}",
+                  payload_remaining_, payload_offset_, n_bytes, Hex::encode(buf, hex_len));
+
+        if (close_received_) {
+          // No data is accepted after CLOSE has been received.
+          ENVOY_LOG(debug, "websocket decoder: data received after CLOSE: {} bytes", n_bytes);
+          buffer_.drain(n_bytes);
+        } else {
+          decoded_.move(buffer_, n_bytes);
+        }
+        payload_remaining_ -= n_bytes;
       }
-      ENVOY_LOG(trace, "websocket decoder: payload remaining: {}, offset: {}, processing: {}: 0x{}",
-                payload_remaining_, payload_offset_, n_bytes, Hex::encode(buf, hex_len));
-
-      decoded_.move(buffer_, n_bytes);
-      payload_remaining_ -= n_bytes;
 
       if (buffer_.length() == 0) {
-        return;
+        return; // continue when there is more data
       }
     }
     //
@@ -837,27 +869,25 @@ void Codec::Decoder::decode(Buffer::Instance& data, bool end_stream) {
 
     uint8_t frame_header[2];
     size_t frame_offset = 0;
-    uint8_t opcode;
-    uint64_t payload_len;
 
     ENVOY_LOG(trace, "websocket decoder: remaining buffer: {} bytes", buffer_.length());
 
     TRY_READ_NETWORK(&frame_header);
-    opcode = frame_header[0] & OPCODE_MASK;
-    payload_len = frame_header[1] & PAYLOAD_LEN_MASK;
+    const uint8_t opcode = frame_header[0] & OPCODE_MASK;
+    const bool final_frame = (frame_header[0] & FIN_MASK) != 0;
+    const bool masked = (frame_header[1] & MASK_MASK) != 0;
+    uint64_t payload_len = frame_header[1] & PAYLOAD_LEN_MASK;
 
     if (payload_len == 126) {
       uint16_t len16;
-
       TRY_READ_NETWORK(&len16);
       payload_len = be16toh(len16);
     } else if (payload_len == 127) {
       uint64_t len64;
-
       TRY_READ_NETWORK(&len64);
       payload_len = be64toh(len64);
     }
-    if (frame_header[1] & MASK_MASK) {
+    if (masked) {
       TRY_READ_NETWORK(&mask_);
       unmasking_ = true;
     }
@@ -866,57 +896,104 @@ void Codec::Decoder::decode(Buffer::Instance& data, bool end_stream) {
     // Whole header received and decoded
     //
 
+    // RFC 6455 section 5.1 requires client to mask all frames it sends to the server,
+    // and prohibits server from ever masking any frames is sends to the client.
+    const bool expected_masked = !parent_.config()->client_;
+    if (masked != expected_masked) {
+      ENVOY_LOG(debug, "websocket decoder: invalid frame masking");
+      goto protocol_error;
+    }
+
+    if (opcode < OPCODE_CLOSE) {
+      // Unframe and forward all non-control frames
+      ENVOY_LOG(trace, "websocket decoder: received websocket data: header {} bytes, data {} bytes",
+                frame_offset, payload_len);
+      buffer_.drain(frame_offset);
+      payload_remaining_ = payload_len;
+      continue; // loop back to frame data decoding
+    }
+
     // Terminate and respond to any control frames
-    if (opcode >= OPCODE_CLOSE) {
-      // Protect against too large control frames that could happen if the decoder ever loses
-      // sync with the data stream.
-      if (payload_len > WEBSOCKET_CONTROL_FRAME_MAX_SIZE) {
-        ENVOY_LOG(debug, "websocket decoder: too large control frame: {} bytes", payload_len);
-        buffer_.drain(buffer_.length());
-        end_stream_ = true;
-        return;
-      }
 
-      // Buffer until whole control frame has been received
-      if (buffer_.length() < frame_offset + payload_len) {
-        return;
-      }
+    // control frames are always final
+    if (!final_frame) {
+      ENVOY_LOG(debug, "websocket decoder: invalid control frame");
+      goto protocol_error;
+    }
 
-      // Drain control frame header, get the payload
-      buffer_.drain(frame_offset);
-      uint8_t* payload = reinterpret_cast<uint8_t*>(buffer_.linearize(payload_len));
+    // Protect against too large control frames that could happen if the decoder ever loses
+    // sync with the data stream.
+    if (payload_len > WEBSOCKET_CONTROL_FRAME_MAX_SIZE) {
+      ENVOY_LOG(debug, "websocket decoder: too large control frame: {} bytes", payload_len);
+      goto protocol_error;
+    }
 
-      // Unmask the control frame payload
-      if (unmasking_) {
-        maskData(payload, payload_len, mask_);
-      }
+    // Buffer until whole control frame has been received
+    if (buffer_.length() < frame_offset + payload_len) {
+      return;
+    }
+
+    // Drain control frame header, get the payload
+    buffer_.drain(frame_offset);
+    uint8_t* payload = reinterpret_cast<uint8_t*>(buffer_.linearize(payload_len));
+
+    // Unmask the control frame payload
+    if (unmasking_) {
+      maskData(payload, payload_len, mask_);
+    }
+
+    // unknown opcodes are ignored, but their data is drained to retain stream sync
+    switch (opcode) {
+    case OPCODE_CLOSE:
+      ENVOY_LOG(trace, "websocket decoder: CLOSE received");
 
-      switch (opcode) {
-      case OPCODE_CLOSE:
-        ENVOY_LOG(trace, "websocket decoder: CLOSE received");
-        end_stream_ = true;
-        break;
-      case OPCODE_PING: {
-        ENVOY_LOG(trace, "websocket decoder: PING received");
-        // Reply with a PONG with the same payload
-        parent_.pong(payload, payload_len);
-        break;
+      // validate CLOSE payload if any
+      if (payload_len > 0) {
+        if (payload_len == 1) {
+          ENVOY_LOG(debug, "websocket decoder: invalid CLOSE payload");
+          goto protocol_error;
+        }
+
+        // we do not interpret the status code to remain compatible with future revisions of the
+        // WebSocket protocol specification, but we validate that any reason is UTF-8 encoded as
+        // required.
+        const absl::string_view reason{reinterpret_cast<const char*>(payload + 2),
+                                       static_cast<size_t>(payload_len - 2)};
+        if (!utf8_range::IsStructurallyValid(reason)) {
+          ENVOY_LOG(debug, "websocket decoder: invalid CLOSE reason");
+          goto protocol_error;
+        }
+
+        // store for sending the frame back
+        if (!close_received_) {
+          close_payload_.assign(reinterpret_cast<const char*>(payload), payload_len);
+        }
       }
-      case OPCODE_PONG:
-        ENVOY_LOG(trace, "websocket decoder: PONG received");
-        break;
+      close_received_ = true;
+      if (parent_.close_sent_) {
+        // Both CLOSE frames have now been exchanged; no later frames need to be processed.
+        buffer_.drain(buffer_.length());
+        return;
       }
-      // Drain control plane payload
-      buffer_.drain(payload_len);
-    } else {
-      // Unframe and forward all non-control frames
-      ENVOY_LOG(trace, "websocket decoder: received websocket data: header {} bytes, data {} bytes",
-                frame_offset, payload_len);
-
-      buffer_.drain(frame_offset);
-      payload_remaining_ = payload_len;
+      break;
+    case OPCODE_PING:
+      ENVOY_LOG(trace, "websocket decoder: PING received");
+      // Reply with a PONG with the same payload
+      parent_.pong(payload, payload_len);
+      break;
+    case OPCODE_PONG:
+      ENVOY_LOG(trace, "websocket decoder: PONG received");
+      break;
     }
+    // Drain control plane payload
+    buffer_.drain(payload_len);
   }
+
+  return;
+
+protocol_error:
+  buffer_.drain(buffer_.length());
+  protocol_error_ = true;
 }
 
 } // namespace WebSocket
diff --git a/cilium/websocket_codec.h b/cilium/websocket_codec.h
index 817940d5f..5739910eb 100644
--- a/cilium/websocket_codec.h
+++ b/cilium/websocket_codec.h
@@ -54,15 +54,13 @@ class Codec : Logger::Loggable<Logger::Id::filter> {
   public:
     Encoder(Codec& parent) : parent_(parent) {}
 
-    void encode(Buffer::Instance&, bool end_stream, uint8_t opcode);
+    void encode(Buffer::Instance&, uint8_t opcode);
 
     size_t hasData() { return encoded_.length() > 0; }
     Buffer::Instance& data() { return encoded_; }
-    bool endStream() { return end_stream_; }
     void drain() { encoded_.drain(encoded_.length()); }
 
     Codec& parent_;
-    bool end_stream_{false};
     Buffer::OwnedImpl encoded_; // Buffer for encoded websocket frames
   };
 
@@ -74,11 +72,12 @@ class Codec : Logger::Loggable<Logger::Id::filter> {
 
     size_t hasData() { return decoded_.length() > 0; }
     Buffer::Instance& data() { return decoded_; }
-    bool endStream() { return end_stream_; }
     void drain() { decoded_.drain(decoded_.length()); }
 
     Codec& parent_;
-    bool end_stream_{false};
+    bool close_received_{false};
+    bool protocol_error_{false};
+    std::string close_payload_;
     Buffer::OwnedImpl buffer_;  // Buffer for partial websocket frames
     Buffer::OwnedImpl decoded_; // Buffer for decoded websocket frames
 
@@ -90,7 +89,7 @@ class Codec : Logger::Loggable<Logger::Id::filter> {
 
   void startPingTimer();
   void resetPingTimer() {
-    if (ping_timer_ != nullptr) {
+    if (ping_timer_ != nullptr && !(close_sent_ && decoder_.close_received_)) {
       auto config = parent_->config();
       if (config->ping_when_idle_) {
         ping_timer_->enableTimer(config->ping_interval_);
@@ -126,6 +125,9 @@ class Codec : Logger::Loggable<Logger::Id::filter> {
   Event::TimerPtr handshake_timer_{nullptr};
   Buffer::OwnedImpl handshake_buffer_;
   bool accepted_{false};
+  bool close_sent_{false};
+  bool decoded_end_stream_sent_{false};
+  bool encoded_end_stream_sent_{false};
 };
 using CodecPtr = std::unique_ptr<Codec>;
 
diff --git a/cilium/websocket_protocol.h b/cilium/websocket_protocol.h
index baf9af5fe..d5b4cbea3 100644
--- a/cilium/websocket_protocol.h
+++ b/cilium/websocket_protocol.h
@@ -4,7 +4,7 @@
 
 // Some sensible limits to protect against excess resource use
 #define WEBSOCKET_HANDSHAKE_MAX_SIZE 4096
-#define WEBSOCKET_CONTROL_FRAME_MAX_SIZE 256
+#define WEBSOCKET_CONTROL_FRAME_MAX_SIZE 125 // RFC 6455 §5.5
 
 /* Ref. RFC 6455 */
 
diff --git a/patches/0009-network-Allow-write-filters-to-consume-data-after-e.patch b/patches/0009-network-Allow-write-filters-to-consume-data-after-e.patch
new file mode 100644
index 000000000..81f1e55e6
--- /dev/null
+++ b/patches/0009-network-Allow-write-filters-to-consume-data-after-e.patch
@@ -0,0 +1,65 @@
+From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
+From: Cilium Authors <dev@cilium.io>
+Date: Mon, 24 Aug 2026 08:00:00 +0200
+Subject: [PATCH] network: Allow write filters to consume data after end stream
+
+A protocol filter may synthesize an end stream for the data it exposes while
+the underlying transport remains open for protocol control frames. Give write
+filters an opportunity to consume such frames after the connection write side
+has been half-closed.
+
+The connection still rejects any data that a filter does not consume, so this
+does not permit application data to be written to the transport after FIN.
+---
+ source/common/network/connection_impl.cc | 22 ++++++++++++----------
+ 1 file changed, 12 insertions(+), 10 deletions(-)
+
+diff --git a/source/common/network/connection_impl.cc b/source/common/network/connection_impl.cc
+index 21a263323b..042f12d65d 100644
+--- a/source/common/network/connection_impl.cc
++++ b/source/common/network/connection_impl.cc
+@@ -596,30 +596,32 @@ void ConnectionImpl::write(Buffer::Instance& data, bool end_stream) {
+ void ConnectionImpl::write(Buffer::Instance& data, bool end_stream, bool through_filter_chain) {
+   ASSERT(!end_stream || enable_half_close_);
+   ASSERT(dispatcher_.isThreadSafe());
+ 
+-  if (write_end_stream_) {
+-    // It is an API violation to write more data after writing end_stream, but a duplicate
+-    // end_stream with no data is harmless. This catches misuse of the API that could result in data
+-    // being lost.
+-    ASSERT(data.length() == 0 && end_stream);
+-
+-    return;
+-  }
+-
+-  if (through_filter_chain) {
++  // A filter may synthesize end_stream for its decoded data while its underlying protocol remains
++  // open for control frames. In that case, give write filters a chance to consume subsequent data.
++  // Any data left unconsumed is still rejected below and cannot reach the transport after FIN.
++  if (through_filter_chain && (!write_end_stream_ || data.length() > 0)) {
+     // NOTE: This is kind of a hack, but currently we don't support restart/continue on the write
+     //       path, so we just pass around the buffer passed to us in this function. If we ever
+     //       support buffer/restart/continue on the write path this needs to get more complicated.
+     current_write_buffer_ = &data;
+     current_write_end_stream_ = end_stream;
+     FilterStatus status = filter_manager_.onWrite();
+     current_write_buffer_ = nullptr;
+
+     if (FilterStatus::StopIteration == status) {
+       return;
+     }
+   }
+ 
++  if (write_end_stream_) {
++    // Any data left after the filters ran is an API violation and cannot be written after FIN. An
++    // empty write is harmless regardless of whether it repeats end_stream.
++    ASSERT(data.length() == 0);
++
++    return;
++  }
++
+   write_end_stream_ = end_stream;
+   if (data.length() > 0 || end_stream) {
+     ENVOY_CONN_LOG(trace, "writing {} bytes, end_stream {}", *this, data.length(), end_stream);
+-- 
+2.43.0
diff --git a/patches/0010-tcp_proxy-Allow-filters-to-wait-for-upstream-close.patch b/patches/0010-tcp_proxy-Allow-filters-to-wait-for-upstream-close.patch
new file mode 100644
index 000000000..e6bf1f8a9
--- /dev/null
+++ b/patches/0010-tcp_proxy-Allow-filters-to-wait-for-upstream-close.patch
@@ -0,0 +1,260 @@
+From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
+From: Cilium Authors <dev@cilium.io>
+Date: Tue, 25 Aug 2026 16:00:00 +0200
+Subject: [PATCH] tcp_proxy: Allow filters to wait for upstream close
+
+A protocol filter may need the upstream transport to remain open after its
+downstream connection has ended so the peer can finish a protocol-level close
+handshake. Add a per-connection filter-state timeout which makes TCP proxy
+flush its pending writes and wait for the upstream peer to close.
+
+Data received during this final drain interval cannot be delivered to the
+closed downstream connection and is discarded. Preserve the existing reset
+behavior for all connections which do not opt in.
+---
+ envoy/tcp/upstream.h                     |  7 ++++++-
+ source/common/tcp_proxy/tcp_proxy.cc     | 31 +++++++++++++++++++++++++------
+ source/common/tcp_proxy/tcp_proxy.h      | 17 +++++++++++++++--
+ source/common/tcp_proxy/upstream.cc      | 17 +++++++++++++----
+ source/common/tcp_proxy/upstream.h       | 12 +++++++++---
+ 5 files changed, 68 insertions(+), 16 deletions(-)
+
+diff --git a/envoy/tcp/upstream.h b/envoy/tcp/upstream.h
+index 9a96694e41..2f79b3bca5 100644
+--- a/envoy/tcp/upstream.h
++++ b/envoy/tcp/upstream.h
+@@ -1,5 +1,7 @@
+ #pragma once
+ 
++#include <chrono>
++
+ #include "envoy/buffer/buffer.h"
+ #include "envoy/extensions/filters/network/tcp_proxy/v3/tcp_proxy.pb.h"
+ #include "envoy/http/filter.h"
+@@ -157,11 +159,14 @@ public:
+    * Called when an event is received on the downstream connection
+    * @param event supplies the event which occurred.
+    * @param details supplies the details of the event, e.g. the local close reason.
++   * @param flush_wait_timeout supplies how long a TCP upstream should wait for its peer to close
++   *        after pending writes have been flushed. Zero preserves the normal immediate close.
+    * @return the underlying ConnectionData if the event is not "Connected" and draining
+              is supported for this upstream.
+    */
+   virtual Tcp::ConnectionPool::ConnectionData*
+-  onDownstreamEvent(Network::ConnectionEvent event, absl::string_view details = "") PURE;
++  onDownstreamEvent(Network::ConnectionEvent event, absl::string_view details = "",
++                    std::chrono::milliseconds flush_wait_timeout = {}) PURE;
+ 
+   /* Called to convert underlying transport socket from non-secure mode
+    * to secure mode. Implemented only by start_tls transport socket.
+diff --git a/source/common/tcp_proxy/tcp_proxy.cc b/source/common/tcp_proxy/tcp_proxy.cc
+index e5d23f2199..405ef9c1bc 100644
+--- a/source/common/tcp_proxy/tcp_proxy.cc
++++ b/source/common/tcp_proxy/tcp_proxy.cc
+@@ -457,6 +457,16 @@ void Filter::initialize(Network::ReadFilterCallbacks& callbacks, bool set_connect
+     }
+   }
+ 
++  const StreamInfo::UInt64Accessor* upstream_flush_wait_timeout =
++      read_callbacks_->connection()
++          .streamInfo()
++          .filterState()
++          ->getDataReadOnly<StreamInfo::UInt64Accessor>(UpstreamFlushWaitTimeoutMs);
++  if (upstream_flush_wait_timeout != nullptr) {
++    upstream_flush_wait_timeout_ =
++        std::chrono::milliseconds(upstream_flush_wait_timeout->value());
++  }
++
+   // Handle TLS handshake wait mode.
+   if (connect_mode_ == UpstreamConnectMode::ON_DOWNSTREAM_TLS_HANDSHAKE) {
+     const auto ssl_connection = read_callbacks_->connection().ssl();
+@@ -1204,12 +1214,14 @@ void Filter::onDownstreamEvent(Network::ConnectionEvent event) {
+   if (upstream_) {
+     absl::string_view downstream_close_details = read_callbacks_->connection().localCloseReason();
+     Tcp::ConnectionPool::ConnectionDataPtr conn_data(
+-        upstream_->onDownstreamEvent(event, downstream_close_details));
++        upstream_->onDownstreamEvent(event, downstream_close_details,
++                                     upstream_flush_wait_timeout_));
+     if (conn_data != nullptr &&
+         conn_data->connection().state() != Network::Connection::State::Closed) {
+       config_->drainManager().add(config_->sharedConfig(), std::move(conn_data),
+                                   std::move(upstream_callbacks_), std::move(idle_timer_),
+-                                  idle_timeout_, read_callbacks_->upstreamHost());
++                                  idle_timeout_, read_callbacks_->upstreamHost(),
++                                  upstream_flush_wait_timeout_.count() > 0);
+     }
+     if (event == Network::ConnectionEvent::LocalClose ||
+         event == Network::ConnectionEvent::RemoteClose) {
+@@ -1508,8 +1520,10 @@ void UpstreamDrainManager::add(const Config::SharedConfigSharedPtr& config,
+                                Event::TimerPtr&& idle_timer,
+                                absl::optional<std::chrono::milliseconds> idle_timeout,
+-                               const Upstream::HostDescriptionConstSharedPtr& upstream_host) {
++                               const Upstream::HostDescriptionConstSharedPtr& upstream_host,
++                               bool discard_upstream_data) {
+   DrainerPtr drainer(new Drainer(*this, config, callbacks, std::move(upstream_conn_data),
+-                                 std::move(idle_timer), idle_timeout, upstream_host));
++                                 std::move(idle_timer), idle_timeout, upstream_host,
++                                 discard_upstream_data));
+   callbacks->drain(*drainer);
+ 
+   // Use temporary to ensure we get the pointer before we move it out of drainer
+@@ -1530,9 +1544,10 @@ Drainer::Drainer(UpstreamDrainManager& parent, const Config::SharedConfigSharedPt
+                  Tcp::ConnectionPool::ConnectionDataPtr&& conn_data, Event::TimerPtr&& idle_timer,
+                  absl::optional<std::chrono::milliseconds> idle_timeout,
+-                 const Upstream::HostDescriptionConstSharedPtr& upstream_host)
++                 const Upstream::HostDescriptionConstSharedPtr& upstream_host,
++                 bool discard_upstream_data)
+     : parent_(parent), callbacks_(callbacks), upstream_conn_data_(std::move(conn_data)),
+       idle_timer_(std::move(idle_timer)), idle_timeout_(idle_timeout),
+-      upstream_host_(upstream_host), config_(config) {
++      upstream_host_(upstream_host), config_(config), discard_upstream_data_(discard_upstream_data) {
+   ENVOY_CONN_LOG(trace, "draining the upstream connection", upstream_conn_data_->connection());
+   config_->stats().upstream_flush_total_.inc();
+   config_->stats().upstream_flush_active_.inc();
+@@ -1551,6 +1564,10 @@ void Drainer::onEvent(Network::ConnectionEvent event) {
+ }
+ 
+ void Drainer::onData(Buffer::Instance& data, bool) {
++  if (discard_upstream_data_) {
++    data.drain(data.length());
++    return;
++  }
+   if (data.length() > 0) {
+     // There is no downstream connection to send any data to, but the upstream
+     // sent some data. Try to behave similar to what the kernel would do
+diff --git a/source/common/tcp_proxy/tcp_proxy.h b/source/common/tcp_proxy/tcp_proxy.h
+index b10971523a..39540726b7 100644
+--- a/source/common/tcp_proxy/tcp_proxy.h
++++ b/source/common/tcp_proxy/tcp_proxy.h
+@@ -55,6 +55,15 @@ constexpr absl::string_view PerConnectionIdleTimeoutMs =
+  */
+ constexpr absl::string_view ReceiveBeforeConnectKey = "envoy.tcp_proxy.receive_before_connect";
+ 
++/**
++ * UpstreamFlushWaitTimeoutMs is the key for a per-connection upstream flush wait timeout. A
++ * non-zero ``StreamInfo::UInt64Accessor`` value makes a raw TCP upstream wait for its peer to
++ * close after pending writes have been flushed when the downstream connection closes. Data
++ * received during this interval is discarded because the downstream connection no longer exists.
++ */
++constexpr absl::string_view UpstreamFlushWaitTimeoutMs =
++    "envoy.tcp_proxy.upstream_flush_wait_timeout_ms";
++
+ /**
+  * All tcp proxy stats. @see stats_macros.h
+  */
+@@ -755,4 +764,5 @@ protected:
+   bool receive_before_connect_{false};
++  std::chrono::milliseconds upstream_flush_wait_timeout_{0};
+   bool early_data_end_stream_{false};
+   Buffer::OwnedImpl early_data_buffer_{};
+   HttpStreamDecoderFilterCallbacks upstream_decoder_filter_callbacks_;
+@@ -778,6 +788,7 @@ public:
+           Tcp::ConnectionPool::ConnectionDataPtr&& conn_data, Event::TimerPtr&& idle_timer,
+           absl::optional<std::chrono::milliseconds> idle_timeout,
+-          const Upstream::HostDescriptionConstSharedPtr& upstream_host);
++          const Upstream::HostDescriptionConstSharedPtr& upstream_host,
++          bool discard_upstream_data);
+ 
+   void onEvent(Network::ConnectionEvent event);
+   void onData(Buffer::Instance& data, bool end_stream);
+@@ -794,5 +805,6 @@ private:
+   Upstream::HostDescriptionConstSharedPtr upstream_host_;
+   Config::SharedConfigSharedPtr config_;
++  const bool discard_upstream_data_;
+ };
+ 
+ using DrainerPtr = std::unique_ptr<Drainer>;
+@@ -806,6 +818,7 @@ public:
+            const std::shared_ptr<Filter::UpstreamCallbacks>& callbacks,
+            Event::TimerPtr&& idle_timer, absl::optional<std::chrono::milliseconds> idle_timeout,
+-           const Upstream::HostDescriptionConstSharedPtr& upstream_host);
++           const Upstream::HostDescriptionConstSharedPtr& upstream_host,
++           bool discard_upstream_data);
+   void remove(Drainer& drainer, Event::Dispatcher& dispatcher);
+ 
+ private:
+diff --git a/source/common/tcp_proxy/upstream.cc b/source/common/tcp_proxy/upstream.cc
+index f466703e25..9d12c690d4 100644
+--- a/source/common/tcp_proxy/upstream.cc
++++ b/source/common/tcp_proxy/upstream.cc
+@@ -125,14 +125,21 @@ StreamInfo::DetectedCloseType TcpUpstream::detectedCloseType() const {
+ }
+ 
+ Tcp::ConnectionPool::ConnectionData* TcpUpstream::onDownstreamEvent(Network::ConnectionEvent event,
+-                                                                    absl::string_view details) {
++                                                                    absl::string_view details,
++                                                                    std::chrono::milliseconds
++                                                                        flush_wait_timeout) {
+   // TODO(botengyao): propagate RST back to upstream connection if RST is received from downstream.
+   if (event == Network::ConnectionEvent::RemoteClose) {
+     // The close call may result in this object being deleted. Latch the
+     // connection locally so it can be returned for potential draining.
+     auto* conn_data = upstream_conn_data_.release();
++    const bool wait_for_peer_close = flush_wait_timeout.count() > 0;
++    if (wait_for_peer_close) {
++      conn_data->connection().setDelayedCloseTimeout(flush_wait_timeout);
++    }
+     conn_data->connection().close(
+-        Network::ConnectionCloseType::FlushWrite,
++        wait_for_peer_close ? Network::ConnectionCloseType::FlushWriteAndDelay
++                            : Network::ConnectionCloseType::FlushWrite,
+         StreamInfo::LocalCloseReasons::get().ClosingUpstreamTcpDueToDownstreamRemoteClose);
+     return conn_data;
+   } else if (event == Network::ConnectionEvent::LocalClose) {
+@@ -235,7 +242,8 @@ void HttpUpstream::addBytesSentCallback(Network::Connection::BytesSentCb) {
+ }
+ 
+ Tcp::ConnectionPool::ConnectionData*
+-HttpUpstream::onDownstreamEvent(Network::ConnectionEvent event, absl::string_view /*details*/) {
++HttpUpstream::onDownstreamEvent(Network::ConnectionEvent event, absl::string_view /*details*/,
++                                std::chrono::milliseconds /*flush_wait_timeout*/) {
+   if (event == Network::ConnectionEvent::LocalClose ||
+       event == Network::ConnectionEvent::RemoteClose) {
+     resetEncoder(Network::ConnectionEvent::LocalClose, false);
+@@ -563,7 +571,8 @@ bool CombinedUpstream::readDisable(bool disable) {
+ }
+ 
+ Tcp::ConnectionPool::ConnectionData*
+-CombinedUpstream::onDownstreamEvent(Network::ConnectionEvent event, absl::string_view /*details*/) {
++CombinedUpstream::onDownstreamEvent(Network::ConnectionEvent event, absl::string_view /*details*/,
++                                    std::chrono::milliseconds /*flush_wait_timeout*/) {
+   if (!upstream_request_) {
+     return nullptr;
+   }
+diff --git a/source/common/tcp_proxy/upstream.h b/source/common/tcp_proxy/upstream.h
+index 4ce796b563..090856885e 100644
+--- a/source/common/tcp_proxy/upstream.h
++++ b/source/common/tcp_proxy/upstream.h
+@@ -177,7 +177,9 @@ public:
+   void encodeData(Buffer::Instance& data, bool end_stream) override;
+   void addBytesSentCallback(Network::Connection::BytesSentCb cb) override;
+   Tcp::ConnectionPool::ConnectionData* onDownstreamEvent(Network::ConnectionEvent event,
+-                                                         absl::string_view details = "") override;
++                                                         absl::string_view details = "",
++                                                         std::chrono::milliseconds
++                                                             flush_wait_timeout = {}) override;
+   bool startUpstreamSecureTransport() override;
+   Ssl::ConnectionInfoConstSharedPtr getUpstreamConnectionSslInfo() override;
+   StreamInfo::DetectedCloseType detectedCloseType() const override;
+@@ -207,7 +209,9 @@ public:
+   void encodeData(Buffer::Instance& data, bool end_stream) override;
+   void addBytesSentCallback(Network::Connection::BytesSentCb cb) override;
+   Tcp::ConnectionPool::ConnectionData* onDownstreamEvent(Network::ConnectionEvent event,
+-                                                         absl::string_view details = "") override;
++                                                         absl::string_view details = "",
++                                                         std::chrono::milliseconds
++                                                             flush_wait_timeout = {}) override;
+   // HTTP upstream must not implement converting upstream transport
+   // socket from non-secure to secure mode.
+   bool startUpstreamSecureTransport() override { return false; }
+@@ -299,7 +303,9 @@ public:
+   void newStream(GenericConnectionPoolCallbacks& callbacks);
+   void encodeData(Buffer::Instance& data, bool end_stream) override;
+   Tcp::ConnectionPool::ConnectionData* onDownstreamEvent(Network::ConnectionEvent event,
+-                                                         absl::string_view details = "") override;
++                                                         absl::string_view details = "",
++                                                         std::chrono::milliseconds
++                                                             flush_wait_timeout = {}) override;
+   bool isValidResponse(const Http::ResponseHeaderMap&);
+   bool readDisable(bool disable) override;
+   void setConnPoolCallbacks(std::unique_ptr<HttpConnPool::Callbacks>&& callbacks) {
diff --git a/tests/cilium_websocket_codec_integration_test.cc b/tests/cilium_websocket_codec_integration_test.cc
index c4d1f70bf..50e619184 100644
--- a/tests/cilium_websocket_codec_integration_test.cc
+++ b/tests/cilium_websocket_codec_integration_test.cc
@@ -7,6 +7,9 @@
 #include <cstdint>
 #include <string>
 
+#include "envoy/common/platform.h"
+#include "envoy/network/connection.h"
+
 #include "test/integration/fake_upstream.h"
 #include "test/integration/integration_tcp_client.h"
 #include "test/test_common/environment.h"
@@ -134,9 +137,22 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamWritesFirst) {
   ASSERT_TRUE(fake_upstream_connection->waitForData(5, &received));
   ASSERT_EQ(received, "hello");
 
-  ASSERT_TRUE(fake_upstream_connection->write("", true));
+  // A FIN in one direction must not prevent data from flowing in the reverse direction. The first
+  // CLOSE crosses both WebSocket codecs and becomes a half-close at the TCP client.
+  ASSERT_TRUE(fake_upstream_connection->write("upstream final", true));
+  tcp_client->waitForData("helloupstream final");
   tcp_client->waitForHalfClose();
-  ASSERT_TRUE(tcp_client->write("", true));
+
+  // The first CLOSE is only a directional FIN. Keepalive PING/PONG processing must continue while
+  // the reverse direction remains open.
+  const uint64_t ping_count = test_server_->counter("websocket.ping_sent_count")->value();
+  test_server_->waitForCounterGe("websocket.ping_sent_count", ping_count + 1);
+
+  // The TCP client can still send after receiving that FIN. Its own FIN produces the CLOSE response
+  // only after the reverse-direction data has crossed the tunnel.
+  ASSERT_TRUE(tcp_client->write("downstream final", true));
+  ASSERT_TRUE(fake_upstream_connection->waitForData(5 + sizeof("downstream final") - 1, &received));
+  ASSERT_EQ(received, "hellodownstream final");
   ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
   ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
 }
@@ -165,6 +181,23 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamDisconnect) {
   EXPECT_EQ("world", tcp_client->data());
 }
 
+#if ENVOY_PLATFORM_ENABLE_SEND_RST
+// A TCP reset is an abort, not a directional FIN, and must tear down the tunnel immediately.
+TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamReset) {
+  initialize();
+  IntegrationTcpClientPtr tcp_client = makeTcpConnection(lookupPort("tcp_proxy"));
+  ASSERT_TRUE(tcp_client->write("hello"));
+  FakeRawConnectionPtr fake_upstream_connection;
+  ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection));
+  ASSERT_TRUE(fake_upstream_connection->waitForData(5));
+
+  ASSERT_TRUE(fake_upstream_connection->close(Network::ConnectionCloseType::AbortReset));
+  ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
+  tcp_client->waitForDisconnect();
+}
+
+#endif
+
 // Test proxying data in both directions, and that all data is flushed properly
 // when the client disconnects.
 TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamDisconnect) {
@@ -186,7 +219,12 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamDisconnect) {
   ASSERT_TRUE(fake_upstream_connection->waitForData(10, &received));
   ASSERT_EQ(received, "hellohello");
   ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
-  ASSERT_TRUE(fake_upstream_connection->write("", true));
+
+  const uint64_t ping_count = test_server_->counter("websocket.ping_sent_count")->value();
+  test_server_->waitForCounterGe("websocket.ping_sent_count", ping_count + 1);
+
+  ASSERT_TRUE(fake_upstream_connection->write("upstream final", true));
+  tcp_client->waitForData("worldupstream final");
   ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
   tcp_client->waitForDisconnect();
 }
@@ -252,8 +290,7 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamFlush) {
   tcp_client->readDisable(true);
   ASSERT_TRUE(tcp_client->write("", true));
 
-  // This ensures that readDisable(true) has been run on it's thread
-  // before tcp_client starts writing.
+  // Confirm that the downstream FIN crossed the WebSocket tunnel before sending a large response.
   ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
 
   ASSERT_TRUE(fake_upstream_connection->write(data, true));
@@ -265,7 +302,6 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamFlush) {
   tcp_client->readDisable(false);
   tcp_client->waitForData(data);
   tcp_client->waitForHalfClose();
-  ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
 
   uint32_t upstream_pauses =
       test_server_->counter("cluster.cluster1.upstream_flow_control_paused_reading_total")->value();
@@ -295,8 +331,7 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlush) {
   ASSERT_TRUE(fake_upstream_connection->readDisable(true));
   ASSERT_TRUE(fake_upstream_connection->write("", true));
 
-  // This ensures that fake_upstream_connection->readDisable has been run on
-  // it's thread before tcp_client starts writing.
+  // Confirm that the upstream FIN crossed the WebSocket tunnel before sending a large request.
   tcp_client->waitForHalfClose();
 
   ASSERT_TRUE(tcp_client->write(data, true, true, std::chrono::milliseconds(30000)));
@@ -309,7 +344,6 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlush) {
   ASSERT_EQ(received, data);
   ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
   ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
-  tcp_client->waitForHalfClose();
 
   EXPECT_EQ(test_server_->counter("tcp.tcp_stats.upstream_flush_total")->value(), 1);
   test_server_->waitForGaugeEq("tcp.tcp_stats.upstream_flush_active", 0);
@@ -329,15 +363,15 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlushEnvoyExit) {
   FakeRawConnectionPtr fake_upstream_connection;
   ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection));
 
+  // Confirm that the WebSocket handshake and keepalive timer are active.
+  test_server_->waitForCounterGe("websocket.ping_sent_count", 1);
+
   ASSERT_TRUE(fake_upstream_connection->readDisable(true));
   ASSERT_TRUE(fake_upstream_connection->write("", true));
 
-  // This ensures that fake_upstream_connection->readDisable has been run on
-  // it's thread before tcp_client starts writing.
+  // Confirm that the upstream FIN crossed the WebSocket tunnel before filling the write buffer.
   tcp_client->waitForHalfClose();
 
-  test_server_->waitForCounterGe("websocket.ping_sent_count", 1);
-
   ASSERT_TRUE(tcp_client->write(data, true));
 
   test_server_->waitForGaugeEq("tcp.tcp_stats.upstream_flush_active", 1);
diff --git a/tests/cilium_websocket_decap_integration_test.cc b/tests/cilium_websocket_decap_integration_test.cc
index 46f9411fe..d02f7adc3 100644
--- a/tests/cilium_websocket_decap_integration_test.cc
+++ b/tests/cilium_websocket_decap_integration_test.cc
@@ -3,7 +3,9 @@
 #include <gtest/gtest-param-test.h>
 #include <gtest/gtest.h>
 
+#include <array>
 #include <cstddef>
+#include <cstdint>
 #include <string>
 
 #include "envoy/event/dispatcher.h"
@@ -23,6 +25,42 @@ using namespace std::literals;
 
 namespace Envoy {
 
+namespace {
+
+enum class PayloadLengthEncoding { Minimal, Uint16, Uint64 };
+
+std::string maskedClientFrame(absl::string_view payload,
+                              PayloadLengthEncoding encoding = PayloadLengthEncoding::Minimal) {
+  constexpr std::array<uint8_t, 4> mask = {0x12, 0x34, 0x56, 0x78};
+  std::string frame;
+  frame.reserve(14 + payload.size());
+  frame.push_back('\x82'); // FIN and binary opcode
+
+  if (encoding == PayloadLengthEncoding::Minimal && payload.size() < 126) {
+    frame.push_back(static_cast<char>(0x80 | payload.size()));
+  } else if (encoding == PayloadLengthEncoding::Uint16 ||
+             (encoding == PayloadLengthEncoding::Minimal && payload.size() <= UINT16_MAX)) {
+    frame.push_back(static_cast<char>(0x80 | 126));
+    frame.push_back(static_cast<char>(payload.size() >> 8));
+    frame.push_back(static_cast<char>(payload.size()));
+  } else {
+    frame.push_back(static_cast<char>(0x80 | 127));
+    for (int shift = 56; shift >= 0; shift -= 8) {
+      frame.push_back(static_cast<char>(payload.size() >> shift));
+    }
+  }
+
+  for (uint8_t byte : mask) {
+    frame.push_back(static_cast<char>(byte));
+  }
+  for (size_t i = 0; i < payload.size(); ++i) {
+    frame.push_back(static_cast<char>(static_cast<uint8_t>(payload[i]) ^ mask[i % mask.size()]));
+  }
+  return frame;
+}
+
+} // namespace
+
 // params: is_ingress ("true", "false")
 const std::string cilium_tcp_proxy_config_fmt = R"EOF(
 admin:
@@ -158,9 +196,8 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) {
 
   auto client_conn = codec_client_->connection();
 
-  // Create websocket framed data & write it on the client connection
-  Buffer::OwnedImpl buf{"\x82\x5"
-                        "hello"};
+  // Create masked WebSocket framed data and write it on the client connection.
+  Buffer::OwnedImpl buf{maskedClientFrame("hello")};
   client_conn->write(buf, false);
   // Run the dispatcher so that the write event is handled
   client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock);
@@ -181,12 +218,9 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) {
 
   // Send multiple frames back-to-back
   ASSERT_EQ(buf.length(), 0);
-  buf.add("\x82\x6"
-          "hello2"
-          "\x82\x7"
-          "hello21"
-          "\x82\x3"
-          "foo");
+  buf.add(maskedClientFrame("hello2"));
+  buf.add(maskedClientFrame("hello21"));
+  buf.add(maskedClientFrame("foo"));
   client_conn->write(buf, false);
   // Run the dispatcher so that the write event is handled
   client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock);
@@ -207,9 +241,7 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) {
   // Officially optimal length formats must be used, but our implementation
   // accepts larger formats with less data, which makes testing easier.
   ASSERT_EQ(buf.length(), 0);
-  absl::string_view frame16{"\x82\x7e\0\x5"
-                            "len16",
-                            9};
+  const std::string frame16 = maskedClientFrame("len16", PayloadLengthEncoding::Uint16);
   buf.add(frame16);
   client_conn->write(buf, false);
   // Run the dispatcher so that the write event is handled
@@ -234,9 +266,7 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) {
   // Officially optimal length formats must be used, but our implementation
   // accepts larger formats with less data, which makes testing easier.
   ASSERT_EQ(buf.length(), 0);
-  absl::string_view frame64{"\x82\x7f\0\0\0\0\0\0\0\x5"
-                            "len64",
-                            15};
+  const std::string frame64 = maskedClientFrame("len64", PayloadLengthEncoding::Uint64);
   buf.add(frame64);
   client_conn->write(buf, false);
   // Run the dispatcher so that the write event is handled
@@ -255,10 +285,11 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) {
 
   // Gaps within a frame
   ASSERT_EQ(buf.length(), 0);
-  buf.add("\x82\x5"
-          "hello"
-          "\x82\xe"
-          "gap ");
+  const std::string hello_frame = maskedClientFrame("hello");
+  const std::string gap_frame = maskedClientFrame("gap in between");
+  buf.add(hello_frame);
+  // Send the second frame's header, masking key, and first four payload bytes.
+  buf.add(gap_frame.substr(0, 10));
   client_conn->write(buf, false);
   // Run the dispatcher so that the write event is handled
   client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock);
@@ -270,9 +301,8 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) {
   ASSERT_TRUE(fake_upstream_connection->write("bar42"));
 
   ASSERT_EQ(buf.length(), 0);
-  buf.add("in between"
-          "\x82\x3"
-          "foo");
+  buf.add(gap_frame.substr(10));
+  buf.add(maskedClientFrame("foo"));
   client_conn->write(buf, false);
   // Run the dispatcher so that the write event is handled
   client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock);
@@ -287,17 +317,10 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) {
                                "bar42");
   response->clearBody();
 
-  // Masked frames
+  // Masked frame
   ASSERT_EQ(buf.length(), 0);
   auto msg = "heello there\r\n"s;
-  unsigned char mask[4] = {0x12, 0x34, 0x56, 0x78};
-  auto masked = msg;
-  for (size_t i = 0; i < msg.length(); i++) {
-    masked[i] = msg[i] ^ mask[i % 4];
-  }
-  buf.add("\x82\x8e");
-  buf.add(mask, 4);
-  buf.add(masked.data(), masked.length());
+  buf.add(maskedClientFrame(msg));
   client_conn->write(buf, false);
   // Run the dispatcher so that the write event is handled
   client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock);
@@ -315,23 +338,18 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) {
                                 "heello there\r\n");
   response->clearBody();
 
-  // 2nd masked frame
+  // Split masked frame
   ASSERT_EQ(buf.length(), 0);
   auto msg2 = "hello there\r\n"s;
-  unsigned char mask2[4] = {0x90, 0xab, 0xcd, 0xef};
-  auto masked2 = msg2;
-  for (size_t i = 0; i < msg2.length(); i++) {
-    masked2[i] = msg2[i] ^ mask2[i % 4];
-  }
-  // Write frame header
-  buf.add("\x82\x8d");
-  buf.add(mask2, 4);
+  const std::string masked2 = maskedClientFrame(msg2);
+  // Write the frame header and masking key.
+  buf.add(masked2.substr(0, 6));
   client_conn->write(buf, false);
   // Run the dispatcher so that the write event is handled
   client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock);
 
   // Write 5 first bytes
-  buf.add(masked2.data(), 5);
+  buf.add(masked2.substr(6, 5));
   client_conn->write(buf, false);
   // Run the dispatcher so that the write event is handled
   client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock);
@@ -341,7 +359,7 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) {
   seen_data_len = data.length();
 
   // Write remaining bytes
-  buf.add(masked2.data() + 5, masked2.length() - 5);
+  buf.add(masked2.substr(11));
   client_conn->write(buf, false);
   // Run the dispatcher so that the write event is handled
   client_conn->dispatcher().run(Event::Dispatcher::RunType::NonBlock);
@@ -370,4 +388,116 @@ TEST_P(CiliumWebSocketIntegrationTest, AcceptedWebSocket) {
   codec_client_->close();
 }
 
+TEST_P(CiliumWebSocketIntegrationTest, UnmaskedClientFrameRejected) {
+  initialize();
+  auto request_headers = Http::TestRequestHeaderMapImpl{
+      {":method", "GET"},
+      {":path", "/"},
+      {":authority", "host"},
+      {"Upgrade", "websocket"},
+      {"Connection", "Upgrade"},
+      {"Origin", "jarno.cilium.rocks"},
+      {"Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ=="},
+      {"Sec-WebSocket-Version", "13"},
+      {"x-request-id", "000000ff-0000-0000-0000-000000000001"},
+      {"x-envoy-original-dst-host", original_dst_address->asString()}};
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  IntegrationStreamDecoderPtr response = codec_client_->makeHeaderOnlyRequest(request_headers);
+  FakeRawConnectionPtr fake_upstream_connection;
+  ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection));
+  response->waitForHeaders();
+  ASSERT_EQ("101", response->headers().getStatusValue());
+
+  // RFC 6455 section 5.1 requires every client-to-server frame to be masked.
+  const uint8_t unmasked_frame[] = {0x82, 0x05, 'h', 'e', 'l', 'l', 'o'};
+  Buffer::OwnedImpl frame_buffer(unmasked_frame, sizeof(unmasked_frame));
+  auto* client_connection = codec_client_->connection();
+  client_connection->write(frame_buffer, false);
+  client_connection->dispatcher().run(Event::Dispatcher::RunType::NonBlock);
+
+  test_server_->waitForCounterGe("websocket.protocol_error", 1);
+  ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
+  ASSERT_TRUE(codec_client_->waitForDisconnect());
+}
+
+TEST_P(CiliumWebSocketIntegrationTest, CloseResponseWaitsForReverseFin) {
+  enableHalfClose(true);
+  initialize();
+  auto request_headers = Http::TestRequestHeaderMapImpl{
+      {":method", "GET"},
+      {":path", "/"},
+      {":authority", "host"},
+      {"Upgrade", "websocket"},
+      {"Connection", "Upgrade"},
+      {"Origin", "jarno.cilium.rocks"},
+      {"Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ=="},
+      {"Sec-WebSocket-Version", "13"},
+      {"x-request-id", "000000ff-0000-0000-0000-000000000001"},
+      {"x-envoy-original-dst-host", original_dst_address->asString()}};
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  IntegrationStreamDecoderPtr response = codec_client_->makeHeaderOnlyRequest(request_headers);
+  FakeRawConnectionPtr fake_upstream_connection;
+  ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection));
+  response->waitForHeaders();
+  ASSERT_EQ("101", response->headers().getStatusValue());
+
+  // A masked CLOSE carrying status 1000 and reason "done".
+  const uint8_t masked_close[] = {0x88, 0x86, 0, 0, 0, 0, 0x03, 0xe8, 'd', 'o', 'n', 'e'};
+  Buffer::OwnedImpl close_buffer(masked_close, sizeof(masked_close));
+  auto* client_connection = codec_client_->connection();
+  client_connection->write(close_buffer, false);
+  client_connection->dispatcher().run(Event::Dispatcher::RunType::NonBlock);
+
+  // The received CLOSE represents a FIN toward the upstream TCP connection, but that connection's
+  // reverse direction remains usable.
+  ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
+  ASSERT_TRUE(fake_upstream_connection->write("reverse", true));
+
+  // The reverse data frame precedes the delayed, unmasked CLOSE response, which echoes the status
+  // and reason from the peer's CLOSE.
+  const uint8_t expected_response[] = {0x82, 0x07, 'r',  'e',  'v', 'e', 'r', 's', 'e',
+                                       0x88, 0x06, 0x03, 0xe8, 'd', 'o', 'n', 'e'};
+  response->waitForBodyData(sizeof(expected_response));
+  ASSERT_EQ(response->body(), absl::string_view(reinterpret_cast<const char*>(expected_response),
+                                                sizeof(expected_response)));
+  ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
+
+  codec_client_->close();
+}
+
+TEST_P(CiliumWebSocketIntegrationTest, InvalidCloseAbortsImmediately) {
+  initialize();
+  auto request_headers = Http::TestRequestHeaderMapImpl{
+      {":method", "GET"},
+      {":path", "/"},
+      {":authority", "host"},
+      {"Upgrade", "websocket"},
+      {"Connection", "Upgrade"},
+      {"Origin", "jarno.cilium.rocks"},
+      {"Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ=="},
+      {"Sec-WebSocket-Version", "13"},
+      {"x-request-id", "000000ff-0000-0000-0000-000000000001"},
+      {"x-envoy-original-dst-host", original_dst_address->asString()}};
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  IntegrationStreamDecoderPtr response = codec_client_->makeHeaderOnlyRequest(request_headers);
+  FakeRawConnectionPtr fake_upstream_connection;
+  ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection));
+  response->waitForHeaders();
+  ASSERT_EQ("101", response->headers().getStatusValue());
+
+  // RFC 6455 forbids a CLOSE payload of exactly one byte. A protocol error aborts both TCP sides
+  // instead of waiting for the upstream FIN used by the normal tunnel half-close path.
+  const uint8_t invalid_masked_close[] = {0x88, 0x81, 0, 0, 0, 0, 0};
+  Buffer::OwnedImpl close_buffer(invalid_masked_close, sizeof(invalid_masked_close));
+  auto* client_connection = codec_client_->connection();
+  client_connection->write(close_buffer, false);
+  client_connection->dispatcher().run(Event::Dispatcher::RunType::NonBlock);
+
+  ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
+  ASSERT_TRUE(codec_client_->waitForDisconnect());
+}
+
 } // namespace Envoy
diff --git a/tests/cilium_websocket_encap_integration_test.cc b/tests/cilium_websocket_encap_integration_test.cc
index eb2fc758d..b0839f7e9 100644
--- a/tests/cilium_websocket_encap_integration_test.cc
+++ b/tests/cilium_websocket_encap_integration_test.cc
@@ -263,16 +263,55 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketHandshakeSuccess) {
   ASSERT_TRUE(frame_offset > 0);
   ASSERT_EQ(received_data.substr(frame_offset, 5), "hello");
 
-  ASSERT_TRUE(fake_upstream_connection->write("\x82\x5"
-                                              "world"));
-  ASSERT_TRUE(fake_upstream_connection->close());
-  ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
+  ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x5world\x88\0", 9}));
   tcp_client->waitForHalfClose();
-  tcp_client->close();
+  ASSERT_TRUE(tcp_client->write("", true));
+  ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
+  ASSERT_TRUE(fake_upstream_connection->write("", true));
+  ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
 
   EXPECT_EQ("world", tcp_client->data());
 }
 
+// Adapted from TcpProxyIntegrationTest.UpstreamConnectModeEarlyDataWithHalfClose: data and FIN
+// arriving before the upstream handshake must be preserved, including reverse data after the FIN.
+TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketHandshakeWithEarlyDataAndFin) {
+  initialize();
+  IntegrationTcpClientPtr tcp_client = makeTcpConnection(lookupPort("tcp_proxy"));
+  ASSERT_TRUE(tcp_client->write("final_data", true));
+  FakeRawConnectionPtr fake_upstream_connection;
+  ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection));
+
+  std::string expected_handshake =
+      fmt::format(fmt::runtime(EXPECTED_HANDSHAKE_FMT), original_dst_address->asString());
+  std::string received_data;
+  ASSERT_TRUE(fake_upstream_connection->waitForData(expected_handshake.length(), &received_data));
+  ASSERT_EQ(normalizeXRequestId(received_data), sizeof(X_REQUEST_ID_VALUE) - 1);
+  ASSERT_EQ(received_data, expected_handshake);
+
+  std::string handshake_response =
+      fmt::format(fmt::runtime(HANDSHAKE_RESPONSE_FMT), "GjgmQ9MzNsn3h7+vuIzY25rbQ9M=");
+  ASSERT_TRUE(fake_upstream_connection->write(handshake_response));
+
+  // A ten-byte masked data frame is followed by the masked CLOSE representing the early FIN.
+  ASSERT_TRUE(
+      fake_upstream_connection->waitForData(expected_handshake.length() + 16 + 6, &received_data));
+  received_data.erase(0, expected_handshake.length());
+  auto frame_offset = unmaskData(received_data.data(), 16);
+  ASSERT_TRUE(frame_offset > 0);
+  ASSERT_EQ(received_data.substr(frame_offset, 10), "final_data");
+  received_data.erase(0, 16);
+  frame_offset = unmaskData(received_data.data(), 6, OPCODE_CLOSE);
+  ASSERT_TRUE(frame_offset > 0);
+
+  EXPECT_FALSE(fake_upstream_connection->waitForHalfClose(std::chrono::milliseconds(100)));
+  ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x8response\x88\0", 12}, true));
+  tcp_client->waitForData("response");
+  tcp_client->waitForHalfClose();
+  ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
+  ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
+}
+
 // Test successful handshake where client does not send any data, and the server side sends data
 // right after the handshake response.
 TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketHandshakeNoData) {
@@ -293,16 +332,129 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketHandshakeNoData) {
       fmt::format(fmt::runtime(HANDSHAKE_RESPONSE_FMT), "GjgmQ9MzNsn3h7+vuIzY25rbQ9M=");
   ASSERT_TRUE(fake_upstream_connection->write(handshake_response));
 
-  ASSERT_TRUE(fake_upstream_connection->write("\x82\x5"
-                                              "world"));
-  ASSERT_TRUE(fake_upstream_connection->close());
-  ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
+  ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x5world\x88\0", 9}));
   tcp_client->waitForHalfClose();
-  tcp_client->close();
+  ASSERT_TRUE(tcp_client->write("", true));
+  ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
+  ASSERT_TRUE(fake_upstream_connection->write("", true));
+  ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
 
   EXPECT_EQ("world", tcp_client->data());
 }
 
+TEST_P(CiliumWebSocketIntegrationTest, ControlFramesAfterSendingClose) {
+  initialize();
+  IntegrationTcpClientPtr tcp_client = makeTcpConnection(lookupPort("tcp_proxy"));
+  FakeRawConnectionPtr fake_upstream_connection;
+  ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection));
+
+  std::string expected_handshake =
+      fmt::format(fmt::runtime(EXPECTED_HANDSHAKE_FMT), original_dst_address->asString());
+  std::string received_data;
+  ASSERT_TRUE(fake_upstream_connection->waitForData(expected_handshake.length(), &received_data));
+
+  std::string handshake_response =
+      fmt::format(fmt::runtime(HANDSHAKE_RESPONSE_FMT), "GjgmQ9MzNsn3h7+vuIzY25rbQ9M=");
+  ASSERT_TRUE(fake_upstream_connection->write(handshake_response));
+
+  // The local FIN sends a masked CLOSE but leaves the WebSocket transport open for reverse data.
+  ASSERT_TRUE(tcp_client->write("", true));
+  ASSERT_TRUE(
+      fake_upstream_connection->waitForData(expected_handshake.length() + 6, &received_data));
+  // Remove the already handled prefix that waitForData() reintroduces on each call.
+  received_data.erase(0, expected_handshake.length());
+  ASSERT_GT(unmaskData(received_data.data(), 6, OPCODE_CLOSE), 0);
+
+  // PING and PONG remain valid control frames after sending CLOSE. The PING response must echo its
+  // payload, while the unsolicited PONG requires no response.
+  ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x89\x04ping\x8a\x04pong", 12}));
+  ASSERT_TRUE(
+      fake_upstream_connection->waitForData(expected_handshake.length() + 6 + 10, &received_data));
+  // Remove the already handled prefix that waitForData() reintroduces on each call.
+  received_data.erase(0, expected_handshake.length() + 6);
+  const size_t frame_offset = unmaskData(received_data.data(), 10, OPCODE_PONG);
+  ASSERT_GT(frame_offset, 0);
+  EXPECT_EQ(received_data.substr(frame_offset), "ping");
+
+  ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x04done\x88\0", 8}, true));
+  tcp_client->waitForData("done");
+  tcp_client->waitForHalfClose();
+  ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
+  ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
+}
+
+TEST_P(CiliumWebSocketIntegrationTest, ControlFramesAfterReceivingClose) {
+  initialize();
+  IntegrationTcpClientPtr tcp_client = makeTcpConnection(lookupPort("tcp_proxy"));
+  FakeRawConnectionPtr fake_upstream_connection;
+  ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection));
+
+  std::string expected_handshake =
+      fmt::format(fmt::runtime(EXPECTED_HANDSHAKE_FMT), original_dst_address->asString());
+  std::string received_data;
+  ASSERT_TRUE(fake_upstream_connection->waitForData(expected_handshake.length(), &received_data));
+
+  std::string handshake_response =
+      fmt::format(fmt::runtime(HANDSHAKE_RESPONSE_FMT), "GjgmQ9MzNsn3h7+vuIzY25rbQ9M=");
+  ASSERT_TRUE(fake_upstream_connection->write(handshake_response));
+
+  // A peer CLOSE half-closes the local TCP receive side, but the reverse direction and WebSocket
+  // control plane remain active.
+  ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x88\0", 2}));
+  tcp_client->waitForHalfClose();
+  // Data frames after CLOSE are prohibited and drained, but must not hide subsequent control
+  // frames in the same read.
+  ASSERT_TRUE(
+      fake_upstream_connection->write(std::string{"\x82\x07ignored\x89\x04ping\x8a\x04pong", 21}));
+  ASSERT_TRUE(
+      fake_upstream_connection->waitForData(expected_handshake.length() + 10, &received_data));
+  // Remove the already handled prefix that waitForData() reintroduces on each call.
+  received_data.erase(0, expected_handshake.length());
+  size_t frame_offset = unmaskData(received_data.data(), 10, OPCODE_PONG);
+  ASSERT_GT(frame_offset, 0);
+  EXPECT_EQ(received_data.substr(frame_offset), "ping");
+
+  // Sending the reverse data and FIN completes the two directional CLOSE messages.
+  ASSERT_TRUE(tcp_client->write("done", true));
+  ASSERT_TRUE(fake_upstream_connection->waitForData(expected_handshake.length() + 10 + 10 + 6,
+                                                    &received_data));
+  // Remove the already handled prefix that waitForData() reintroduces on each call.
+  received_data.erase(0, expected_handshake.length() + 10);
+  frame_offset = unmaskData(received_data.data(), 10);
+  ASSERT_GT(frame_offset, 0);
+  EXPECT_EQ(received_data.substr(frame_offset, 4), "done");
+  // Remove the already handled prefix that waitForData() reintroduces on each call.
+  received_data.erase(0, 10);
+  ASSERT_GT(unmaskData(received_data.data(), 6, OPCODE_CLOSE), 0);
+
+  ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
+  ASSERT_TRUE(fake_upstream_connection->write("", true));
+  ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
+}
+
+TEST_P(CiliumWebSocketIntegrationTest, WebSocketTransportFinWithoutCloseAborts) {
+  initialize();
+  IntegrationTcpClientPtr tcp_client = makeTcpConnection(lookupPort("tcp_proxy"));
+  FakeRawConnectionPtr fake_upstream_connection;
+  ASSERT_TRUE(fake_upstreams_[0]->waitForRawConnection(fake_upstream_connection));
+
+  std::string expected_handshake =
+      fmt::format(fmt::runtime(EXPECTED_HANDSHAKE_FMT), original_dst_address->asString());
+  ASSERT_TRUE(fake_upstream_connection->waitForData(expected_handshake.length()));
+
+  std::string handshake_response =
+      fmt::format(fmt::runtime(HANDSHAKE_RESPONSE_FMT), "GjgmQ9MzNsn3h7+vuIzY25rbQ9M=");
+  ASSERT_TRUE(fake_upstream_connection->write(handshake_response));
+
+  // An outer transport FIN without a WebSocket CLOSE is an abort, not a tunneled TCP FIN.
+  ASSERT_TRUE(fake_upstream_connection->write("", true));
+  // The raw downstream observes EOF under half-close even though the WebSocket transport was
+  // aborted immediately.
+  tcp_client->waitForHalfClose();
+  ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
+  tcp_client->close();
+}
+
 // Test proxying data in both directions, and that all data is flushed properly
 // when the client disconnects.
 TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamDisconnect) {
@@ -350,8 +502,15 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamDisconnect) {
   ASSERT_TRUE(frame_offset > 0);
   ASSERT_EQ(frame_offset, 6);
 
+  // CLOSE represents the downstream TCP FIN, but must not close the WebSocket transport before
+  // reverse-direction data and its FIN have arrived.
+  EXPECT_FALSE(fake_upstream_connection->waitForHalfClose(std::chrono::milliseconds(100)));
+
+  // The fake WebSocket server sends its final data and CLOSE, then ends the transport as
+  // recommended by RFC 6455 section 7.1.1.
+  ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x4last\x88\0", 8}, true));
+  tcp_client->waitForData("worldlast");
   ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
-  ASSERT_TRUE(fake_upstream_connection->write("", true));
   ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
   tcp_client->waitForDisconnect();
 }
@@ -396,9 +555,13 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketLargeWrite) {
   ASSERT_TRUE(fake_upstream_connection->write("\x82\x7e\x80\x00"s));
   ASSERT_TRUE(fake_upstream_connection->write(data));
   tcp_client->waitForData(data);
-  tcp_client->close();
+  ASSERT_TRUE(tcp_client->write("", true));
+  ASSERT_TRUE(fake_upstream_connection->waitForData(
+      expected_handshake.length() + 2 * 8 + data.size() + 6, &received_data));
+  ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x88\0", 2}));
+  tcp_client->waitForHalfClose();
   ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
-  ASSERT_TRUE(fake_upstream_connection->close());
+  ASSERT_TRUE(fake_upstream_connection->write("", true));
   ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
 
   uint32_t upstream_pauses =
@@ -443,14 +606,18 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamFlush) {
   tcp_client->readDisable(true);
   ASSERT_TRUE(tcp_client->write("", true));
 
-  // This ensures that readDisable(true) has been run on it's thread
-  // before tcp_client starts writing.
-  ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
+  // The downstream FIN is encoded as a masked CLOSE. Unlike a transport half-close, observing this
+  // frame does not prevent the WebSocket peer from sending a large response.
+  ASSERT_TRUE(
+      fake_upstream_connection->waitForData(expected_handshake.length() + 6, &received_data));
+  received_data.erase(0, expected_handshake.length());
+  auto frame_offset = unmaskData(received_data.data(), received_data.length(), OPCODE_CLOSE);
+  ASSERT_TRUE(frame_offset > 0);
 
   // writing data in one large chunk
-
-  ASSERT_TRUE(fake_upstream_connection->write("\x82\x7f\x03\x20\0\0"s));
-  ASSERT_TRUE(fake_upstream_connection->write(data, true));
+  ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x82\x7f\0\0\0\0\x03\x20\0\0", 10}));
+  ASSERT_TRUE(fake_upstream_connection->write(data));
+  ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x88\0", 2}));
 
   test_server_->waitForCounterGe("cluster.cluster1.upstream_flow_control_paused_reading_total", 1);
   EXPECT_EQ(test_server_->counter("cluster.cluster1.upstream_flow_control_resumed_reading_total")
@@ -460,6 +627,8 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamFlush) {
   tcp_client->waitForData(data);
   tcp_client->waitForHalfClose();
   ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
+  ASSERT_TRUE(fake_upstream_connection->write("", true));
+  ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
 
   uint32_t upstream_pauses =
       test_server_->counter("cluster.cluster1.upstream_flow_control_paused_reading_total")->value();
@@ -470,12 +639,12 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamFlush) {
   EXPECT_GT(upstream_resumes, 0);
 }
 
-// Test that an upstream flush works correctly (all data is flushed)
+// Test that an upstream flush works correctly after the upstream direction has half-closed.
 TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlush) {
-  // Use a very large size to make sure it is larger than the kernel socket read
-  // buffer.
+  // Keep the payload larger than the kernel socket read buffer so that an upstream flush is needed,
+  // while leaving buffer-limit headroom for WebSocket framing and control traffic.
   const uint32_t size = 50 * 1024 * 1024;
-  config_helper_.setBufferLimits(size, size);
+  config_helper_.setBufferLimits(2 * size, 2 * size);
   initialize();
 
   std::string data(size, 'a');
@@ -496,33 +665,34 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlush) {
   ASSERT_TRUE(fake_upstream_connection->write(handshake_response));
 
   ASSERT_TRUE(fake_upstream_connection->readDisable(true));
-  ASSERT_TRUE(fake_upstream_connection->write("", true));
+  ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x88\0", 2}));
 
-  // This ensures that fake_upstream_connection->readDisable has been run on
-  // it's thread before tcp_client starts writing.
+  // The peer's CLOSE becomes an upstream FIN, while the downstream-to-upstream direction remains
+  // available for the large request.
   tcp_client->waitForHalfClose();
 
   ASSERT_TRUE(tcp_client->write(data, true, true, std::chrono::milliseconds(30000)));
 
+  test_server_->waitForGaugeEq("tcp.tcp_stats.upstream_flush_active", 1);
+
   ASSERT_TRUE(fake_upstream_connection->readDisable(false));
   size_t min_size = expected_handshake.length() + data.size() + 14 + 6;
   ASSERT_TRUE(
       fake_upstream_connection->waitForData(FakeRawConnection::waitForAtLeastBytes(min_size)));
   ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
+  ASSERT_TRUE(fake_upstream_connection->write("", true));
   ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
-  tcp_client->waitForHalfClose();
 
   test_server_->waitForGaugeEq("tcp.tcp_stats.upstream_flush_active", 0);
   EXPECT_EQ(test_server_->counter("tcp.tcp_stats.upstream_flush_total")->value(), 1);
 }
 
-// Test that Envoy doesn't crash or assert when shutting down with an upstream
-// flush active
+// Test that Envoy doesn't crash or assert when shutting down with an upstream flush active.
 TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlushEnvoyExit) {
-  // Use a very large size to make sure it is larger than the kernel socket read
-  // buffer.
+  // Keep the payload larger than the kernel socket read buffer so that an upstream flush is needed,
+  // while leaving buffer-limit headroom for WebSocket framing and control traffic.
   const uint32_t size = 50 * 1024 * 1024;
-  config_helper_.setBufferLimits(size, size);
+  config_helper_.setBufferLimits(2 * size, 2 * size);
   initialize();
 
   std::string data(size, 'a');
@@ -543,16 +713,12 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamFlushEnvoyExit) {
   ASSERT_TRUE(fake_upstream_connection->write(handshake_response));
 
   ASSERT_TRUE(fake_upstream_connection->readDisable(true));
-  ASSERT_TRUE(fake_upstream_connection->write("", true));
-
-  // This ensures that fake_upstream_connection->readDisable has been run on
-  // it's thread before tcp_client starts writing.
+  ASSERT_TRUE(fake_upstream_connection->write(std::string{"\x88\0", 2}));
   tcp_client->waitForHalfClose();
 
   ASSERT_TRUE(tcp_client->write(data, true));
 
-  // test_server_->waitForCounterGe("tcp.tcp_stats.upstream_flush_total", 1);
-
+  test_server_->waitForGaugeEq("tcp.tcp_stats.upstream_flush_active", 1);
   test_server_.reset();
   ASSERT_TRUE(fake_upstream_connection->close());
   ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
diff --git a/tests/cilium_websocket_policy_integration_test.cc b/tests/cilium_websocket_policy_integration_test.cc
index 17b682c5a..0fdddb72e 100644
--- a/tests/cilium_websocket_policy_integration_test.cc
+++ b/tests/cilium_websocket_policy_integration_test.cc
@@ -278,9 +278,12 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketUpstreamWritesFirst) {
   ASSERT_TRUE(fake_upstream_connection->waitForData(5, &received));
   ASSERT_EQ(received, "hello");
 
-  ASSERT_TRUE(fake_upstream_connection->write("", true));
+  ASSERT_TRUE(fake_upstream_connection->write("upstream final", true));
+  tcp_client->waitForData("helloupstream final");
   tcp_client->waitForHalfClose();
-  ASSERT_TRUE(tcp_client->write("", true));
+  ASSERT_TRUE(tcp_client->write("downstream final", true));
+  ASSERT_TRUE(fake_upstream_connection->waitForData(5 + sizeof("downstream final") - 1, &received));
+  ASSERT_EQ(received, "hellodownstream final");
   ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
   ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
 }
@@ -326,7 +329,8 @@ TEST_P(CiliumWebSocketIntegrationTest, CiliumWebSocketDownstreamDisconnect) {
   ASSERT_TRUE(fake_upstream_connection->waitForData(10, &received));
   ASSERT_EQ(received, "hellohello");
   ASSERT_TRUE(fake_upstream_connection->waitForHalfClose());
-  ASSERT_TRUE(fake_upstream_connection->write("", true));
+  ASSERT_TRUE(fake_upstream_connection->write("upstream final", true));
+  tcp_client->waitForData("worldupstream final");
   ASSERT_TRUE(fake_upstream_connection->waitForDisconnect());
   tcp_client->waitForDisconnect();
 }