[MG_Remote, MG_Test, ci] (P7): a TCP session's data connection is bound to its authenticated control connection by a 128-bit Welcome.dataNonce instead of by arrival order, so a reordered or stale data connection is refused by name and the wire fingerprint moves with the control schema

MobileGL/MG_Remote/CapsCodec.cpp

@@ -476,6 +476,7 @@ namespace MobileGL::MG_Remote {
         inputs.ProgramArtifactsCodecVersion = MG_State::GLState::kProgramArtifactsCodecVersion;
         inputs.ProgramArtifactsSchema = MG_State::GLState::ProgramArtifactsSchemaFingerprint();
         inputs.OpCount = static_cast<Uint64>(MG_Pipe::MGPWireOp::kOpCount);
+        inputs.ControlSchemaRevision = MOBILEGL_PROTOCOL_CONTROL_REVISION;
         inputs.AbiVersion = MOBILEGL_ABI_VERSION(MOBILEGL_PROTOCOL_ABI_MAJOR, MOBILEGL_PROTOCOL_ABI_MINOR);
         inputs.PointerBits = sizeof(void*) * 8;
         const Uint32 endian = 1;

MobileGL/MG_Remote/Client/ClientSession.cpp

@@ -777,8 +777,11 @@ namespace MobileGL::MG_Remote::Client {
             m_connectDial = true;
             const auto endpoint = tcp ? configured : configured.substr(5);
             std::unique_ptr<Transport::SocketTransport> socket;
-            const auto connected = Transport::SocketTransport::ConnectTo(endpoint, kSpawnConnectTimeoutMs, socket);
+            // PH-7 (4): on TCP, the control connection only. The data connection is opened after
+            // Welcome, with the nonce Welcome carries - never paired by arrival order.
+            const auto connected = Transport::SocketTransport::ConnectControl(endpoint, kSpawnConnectTimeoutMs, socket);
             if (connected != MOBILEGL_OK) return connected;
+            m_dataEndpoint = tcp ? endpoint : std::string();
             const auto started = StartOverSocket(std::move(socket));
             if (started == MOBILEGL_OK) {
                 MGLOG_I("MG_Remote client: control=%s data=%s server=%s pid=%u dial=connect",
@@ -890,6 +893,7 @@ namespace MobileGL::MG_Remote::Client {
         // every lane (CONTRACT-P6 4.4).
         Transport::SessionSegmentSizes negotiatedSizes;
         bool stream = false;
+        std::vector<Uint8> dataBind;
         Uint64 announced[4] = {0, 0, 0, 0};
         const Uint32 replySlotCount = Transport::kDefaultReplySlotCount;
         {
@@ -949,6 +953,19 @@ namespace MobileGL::MG_Remote::Client {
                 return result;
             }
             stream = terms->dataPlane() == ::MobileGL::Wire::DataPlane::Stream;
+            if (stream) {
+                // PH-7 (4). A Stream Welcome without a nonce of exactly the minted width names no
+                // data connection this client could open; refused rather than guessed at.
+                const auto* nonce = welcome->dataNonce();
+                if (nonce == nullptr || nonce->size() != Transport::kDataNonceBytes || m_dataEndpoint.empty()) {
+                    const auto result = RefuseHandshake(*m_transport, ::MobileGL::Wire::RefuseCode::LinkTerms,
+                                                        "stream Welcome carries no data nonce", Transport::kDataNonceBytes,
+                                                        nonce == nullptr ? 0 : nonce->size());
+                    Stop();
+                    return result;
+                }
+                dataBind = EncodeDataBind(nonce->data(), nonce->size());
+            }
             negotiatedSizes.CmdRingBytes = terms->cmdWindowBytes();
             negotiatedSizes.StageBytes = terms->stageWindowBytes();
             negotiatedSizes.EventRingBytes = terms->eventWindowBytes();
@@ -969,7 +986,16 @@ namespace MobileGL::MG_Remote::Client {
         }
 
         if (stream) {
-            const auto attached = AttachStreamLink(m_socketTransport->TakeDataFd(), negotiatedSizes);
+            int dataFd = -1;
+            const auto opened = Transport::SocketTransport::ConnectDataConnection(
+                m_dataEndpoint, kSpawnConnectTimeoutMs, MobileGLByteSpan{dataBind.data(), dataBind.size()}, &dataFd);
+            if (opened != MOBILEGL_OK) {
+                MGLOG_E("MG_Remote client: could not open the data connection to %s (rc=%d)",
+                        m_dataEndpoint.c_str(), static_cast<int>(opened));
+                Stop();
+                return opened;
+            }
+            const auto attached = AttachStreamLink(dataFd, negotiatedSizes);
             if (attached != MOBILEGL_OK) { Stop(); return attached; }
             m_controlInbox = std::make_unique<Transport::ControlInbox>(*m_transport);
             Server::ServerLoopInstance().SetRemoteControlSink(&RemoteControlSinkThunk, this);

MobileGL/MG_Remote/Client/ClientSession.h

@@ -450,6 +450,9 @@ namespace MobileGL::MG_Remote::Client {
         Server::LaunchedServer m_spawned;
 
         std::unique_ptr<Transport::SocketTransport> m_socketTransport;
+        // PH-7 (4): the `tcp://` endpoint the control connection went to, for the data
+        // connection opened after Welcome. Empty on every other transport.
+        std::string m_dataEndpoint;
         std::unique_ptr<Transport::Doorbell> m_socketSelfBell;
         std::unique_ptr<Transport::Doorbell> m_socketPeerBell;
 

MobileGL/MG_Remote/Handshake.h

@@ -6,8 +6,10 @@
 #include "Transport/ITransport.h"
 #include <MG_Util/Debug/Log.h>
 #include <cstddef>
+#include <cstdint>
 #include <cstdlib>
 #include <cstring>
+#include <vector>
 
 namespace MobileGL::MG_Remote {
     inline MobileGLResult RefuseHandshake(Transport::ITransport& transport,
@@ -92,4 +94,35 @@ namespace MobileGL::MG_Remote {
                     BuildFingerprint(), stamp == nullptr ? "<missing>" : stamp);
         return MOBILEGL_OK;
     }
+
+    // PH-7 (4), ID-P7-3. THE DATA CONNECTION'S FIRST FRAME, both directions of it.
+    //
+    // A TCP data connection is opened after Welcome and names its session by handing back
+    // `Welcome.dataNonce` in a DataBind, framed exactly like a control message so the server can
+    // tell it from a Hello with one read. The nonce is COMPARED in exactly one place,
+    // ServerSession::BindDataConnection, in constant time; these two only build and parse.
+    inline std::vector<std::uint8_t> EncodeDataBind(const std::uint8_t* nonce, std::size_t size) {
+        ::flatbuffers::FlatBufferBuilder builder(64);
+        auto bind = ::MobileGL::Wire::CreateDataBind(builder, builder.CreateVector(nonce, size));
+        auto envelope = ::MobileGL::Wire::CreateCtrlEnvelope(builder, ::MobileGL::Wire::CtrlMsg::DataBind,
+                                                            bind.Union());
+        ::MobileGL::Wire::FinishCtrlEnvelopeBuffer(builder, envelope);
+        return std::vector<std::uint8_t>(builder.GetBufferPointer(),
+                                         builder.GetBufferPointer() + builder.GetSize());
+    }
+
+    // True when `frame` is a verifiable CtrlEnvelope carrying a DataBind whose nonce is exactly
+    // kDataNonceBytes; the nonce is copied out. Anything else - a Hello, a different schema, a
+    // nonce of another width - is false, and the caller refuses the connection by name.
+    inline bool DecodeDataBind(const std::vector<std::uint8_t>& frame,
+                               std::uint8_t (&nonce)[Transport::kDataNonceBytes]) {
+        if (frame.size() < 8 || !::MobileGL::Wire::CtrlEnvelopeBufferHasIdentifier(frame.data())) return false;
+        ::flatbuffers::Verifier verifier(frame.data(), frame.size());
+        if (!::MobileGL::Wire::VerifyCtrlEnvelopeBuffer(verifier)) return false;
+        const auto* bind = ::MobileGL::Wire::GetCtrlEnvelope(frame.data())->msg_as_DataBind();
+        if (bind == nullptr || bind->nonce() == nullptr || bind->nonce()->size() != Transport::kDataNonceBytes)
+            return false;
+        std::memcpy(nonce, bind->nonce()->data(), Transport::kDataNonceBytes);
+        return true;
+    }
 } // namespace MobileGL::MG_Remote

MobileGL/MG_Remote/Protocol/generated/protocol_generated.h

@@ -47,6 +47,9 @@ struct HelloBuilder;
 struct Welcome;
 struct WelcomeBuilder;
 
+struct DataBind;
+struct DataBindBuilder;
+
 struct CapsSnapshot;
 struct CapsSnapshotBuilder;
 
@@ -502,11 +505,12 @@ enum class CtrlMsg : uint8_t {
   LogLine = 10,
   Refuse = 11,
   LogFlush = 12,
+  DataBind = 13,
   MIN = NONE,
-  MAX = LogFlush
+  MAX = DataBind
 };
 
-inline const CtrlMsg (&EnumValuesCtrlMsg())[13] {
+inline const CtrlMsg (&EnumValuesCtrlMsg())[14] {
   static const CtrlMsg values[] = {
     CtrlMsg::NONE,
     CtrlMsg::Hello,
@@ -520,13 +524,14 @@ inline const CtrlMsg (&EnumValuesCtrlMsg())[13] {
     CtrlMsg::Fatal,
     CtrlMsg::LogLine,
     CtrlMsg::Refuse,
-    CtrlMsg::LogFlush
+    CtrlMsg::LogFlush,
+    CtrlMsg::DataBind
   };
   return values;
 }
 
 inline const char * const *EnumNamesCtrlMsg() {
-  static const char * const names[14] = {
+  static const char * const names[15] = {
     "NONE",
     "Hello",
     "Welcome",
@@ -540,13 +545,14 @@ inline const char * const *EnumNamesCtrlMsg() {
     "LogLine",
     "Refuse",
     "LogFlush",
+    "DataBind",
     nullptr
   };
   return names;
 }
 
 inline const char *EnumNameCtrlMsg(CtrlMsg e) {
-  if (::flatbuffers::IsOutRange(e, CtrlMsg::NONE, CtrlMsg::LogFlush)) return "";
+  if (::flatbuffers::IsOutRange(e, CtrlMsg::NONE, CtrlMsg::DataBind)) return "";
   const size_t index = static_cast<size_t>(e);
   return EnumNamesCtrlMsg()[index];
 }
@@ -603,6 +609,10 @@ template<> struct CtrlMsgTraits<MobileGL::Wire::LogFlush> {
   static const CtrlMsg enum_value = CtrlMsg::LogFlush;
 };
 
+template<> struct CtrlMsgTraits<MobileGL::Wire::DataBind> {
+  static const CtrlMsg enum_value = CtrlMsg::DataBind;
+};
+
 template <bool B = false>
 bool VerifyCtrlMsg(::flatbuffers::VerifierTemplate<B> &verifier, const void *obj, CtrlMsg type);
 template <bool B = false>
@@ -1164,7 +1174,8 @@ struct Welcome FLATBUFFERS_FINAL_CLASS : private ::flatbuffers::Table {
     VT_ABIFINGERPRINT = 20,
     VT_WIREFINGERPRINT = 22,
     VT_LINKTERMS = 24,
-    VT_BACKENDTYPE = 26
+    VT_BACKENDTYPE = 26,
+    VT_DATANONCE = 28
   };
   uint32_t abiMajor() const {
     return GetField<uint32_t>(VT_ABIMAJOR, 0);
@@ -1202,6 +1213,9 @@ struct Welcome FLATBUFFERS_FINAL_CLASS : private ::flatbuffers::Table {
   uint32_t backendType() const {
     return GetField<uint32_t>(VT_BACKENDTYPE, 0);
   }
+  const ::flatbuffers::Vector<uint8_t> *dataNonce() const {
+    return GetPointer<const ::flatbuffers::Vector<uint8_t> *>(VT_DATANONCE);
+  }
   template <bool B = false>
   bool Verify(::flatbuffers::VerifierTemplate<B> &verifier) const {
     return VerifyTableStart(verifier) &&
@@ -1223,6 +1237,8 @@ struct Welcome FLATBUFFERS_FINAL_CLASS : private ::flatbuffers::Table {
            VerifyOffset(verifier, VT_LINKTERMS) &&
            verifier.VerifyTable(linkTerms()) &&
            VerifyField<uint32_t>(verifier, VT_BACKENDTYPE, 4) &&
+           VerifyOffset(verifier, VT_DATANONCE) &&
+           verifier.VerifyVector(dataNonce()) &&
            verifier.EndTable();
   }
 };
@@ -1267,6 +1283,9 @@ struct WelcomeBuilder {
   void add_backendType(uint32_t backendType) {
     fbb_.AddElement<uint32_t>(Welcome::VT_BACKENDTYPE, backendType, 0);
   }
+  void add_dataNonce(::flatbuffers::Offset<::flatbuffers::Vector<uint8_t>> dataNonce) {
+    fbb_.AddOffset(Welcome::VT_DATANONCE, dataNonce);
+  }
   explicit WelcomeBuilder(::flatbuffers::FlatBufferBuilder &_fbb)
         : fbb_(_fbb) {
     start_ = fbb_.StartTable();
@@ -1291,10 +1310,12 @@ inline ::flatbuffers::Offset<Welcome> CreateWelcome(
     uint64_t abiFingerprint = 0,
     uint64_t wireFingerprint = 0,
     ::flatbuffers::Offset<MobileGL::Wire::LinkTerms> linkTerms = 0,
-    uint32_t backendType = 0) {
+    uint32_t backendType = 0,
+    ::flatbuffers::Offset<::flatbuffers::Vector<uint8_t>> dataNonce = 0) {
   WelcomeBuilder builder_(_fbb);
   builder_.add_wireFingerprint(wireFingerprint);
   builder_.add_abiFingerprint(abiFingerprint);
+  builder_.add_dataNonce(dataNonce);
   builder_.add_backendType(backendType);
   builder_.add_linkTerms(linkTerms);
   builder_.add_buildFingerprint(buildFingerprint);
@@ -1326,8 +1347,10 @@ inline ::flatbuffers::Offset<Welcome> CreateWelcomeDirect(
     uint64_t abiFingerprint = 0,
     uint64_t wireFingerprint = 0,
     ::flatbuffers::Offset<MobileGL::Wire::LinkTerms> linkTerms = 0,
-    uint32_t backendType = 0) {
+    uint32_t backendType = 0,
+    const std::vector<uint8_t> *dataNonce = nullptr) {
   auto buildFingerprint__ = buildFingerprint ? _fbb.CreateString(buildFingerprint) : 0;
+  auto dataNonce__ = dataNonce ? _fbb.CreateVector<uint8_t>(*dataNonce) : 0;
   return MobileGL::Wire::CreateWelcome(
       _fbb,
       abiMajor,
@@ -1341,7 +1364,66 @@ inline ::flatbuffers::Offset<Welcome> CreateWelcomeDirect(
       abiFingerprint,
       wireFingerprint,
       linkTerms,
-      backendType);
+      backendType,
+      dataNonce__);
+}
+
+struct DataBind FLATBUFFERS_FINAL_CLASS : private ::flatbuffers::Table {
+  typedef DataBindBuilder Builder;
+  struct Traits;
+  enum FlatBuffersVTableOffset FLATBUFFERS_VTABLE_UNDERLYING_TYPE {
+    VT_NONCE = 4
+  };
+  const ::flatbuffers::Vector<uint8_t> *nonce() const {
+    return GetPointer<const ::flatbuffers::Vector<uint8_t> *>(VT_NONCE);
+  }
+  template <bool B = false>
+  bool Verify(::flatbuffers::VerifierTemplate<B> &verifier) const {
+    return VerifyTableStart(verifier) &&
+           VerifyOffset(verifier, VT_NONCE) &&
+           verifier.VerifyVector(nonce()) &&
+           verifier.EndTable();
+  }
+};
+
+struct DataBindBuilder {
+  typedef DataBind Table;
+  ::flatbuffers::FlatBufferBuilder &fbb_;
+  ::flatbuffers::uoffset_t start_;
+  void add_nonce(::flatbuffers::Offset<::flatbuffers::Vector<uint8_t>> nonce) {
+    fbb_.AddOffset(DataBind::VT_NONCE, nonce);
+  }
+  explicit DataBindBuilder(::flatbuffers::FlatBufferBuilder &_fbb)
+        : fbb_(_fbb) {
+    start_ = fbb_.StartTable();
+  }
+  ::flatbuffers::Offset<DataBind> Finish() {
+    const auto end = fbb_.EndTable(start_);
+    auto o = ::flatbuffers::Offset<DataBind>(end);
+    return o;
+  }
+};
+
+inline ::flatbuffers::Offset<DataBind> CreateDataBind(
+    ::flatbuffers::FlatBufferBuilder &_fbb,
+    ::flatbuffers::Offset<::flatbuffers::Vector<uint8_t>> nonce = 0) {
+  DataBindBuilder builder_(_fbb);
+  builder_.add_nonce(nonce);
+  return builder_.Finish();
+}
+
+struct DataBind::Traits {
+  using type = DataBind;
+  static auto constexpr Create = CreateDataBind;
+};
+
+inline ::flatbuffers::Offset<DataBind> CreateDataBindDirect(
+    ::flatbuffers::FlatBufferBuilder &_fbb,
+    const std::vector<uint8_t> *nonce = nullptr) {
+  auto nonce__ = nonce ? _fbb.CreateVector<uint8_t>(*nonce) : 0;
+  return MobileGL::Wire::CreateDataBind(
+      _fbb,
+      nonce__);
 }
 
 struct CapsSnapshot FLATBUFFERS_FINAL_CLASS : private ::flatbuffers::Table {
@@ -2188,6 +2270,9 @@ struct CtrlEnvelope FLATBUFFERS_FINAL_CLASS : private ::flatbuffers::Table {
   const MobileGL::Wire::LogFlush *msg_as_LogFlush() const {
     return msg_type() == MobileGL::Wire::CtrlMsg::LogFlush ? static_cast<const MobileGL::Wire::LogFlush *>(msg()) : nullptr;
   }
+  const MobileGL::Wire::DataBind *msg_as_DataBind() const {
+    return msg_type() == MobileGL::Wire::CtrlMsg::DataBind ? static_cast<const MobileGL::Wire::DataBind *>(msg()) : nullptr;
+  }
   template <bool B = false>
   bool Verify(::flatbuffers::VerifierTemplate<B> &verifier) const {
     return VerifyTableStart(verifier) &&
@@ -2246,6 +2331,10 @@ template<> inline const MobileGL::Wire::LogFlush *CtrlEnvelope::msg_as<MobileGL:
   return msg_as_LogFlush();
 }
 
+template<> inline const MobileGL::Wire::DataBind *CtrlEnvelope::msg_as<MobileGL::Wire::DataBind>() const {
+  return msg_as_DataBind();
+}
+
 struct CtrlEnvelopeBuilder {
   typedef CtrlEnvelope Table;
   ::flatbuffers::FlatBufferBuilder &fbb_;
@@ -2336,6 +2425,10 @@ inline bool VerifyCtrlMsg(::flatbuffers::VerifierTemplate<B> &verifier, const vo
       auto ptr = reinterpret_cast<const MobileGL::Wire::LogFlush *>(obj);
       return verifier.VerifyTable(ptr);
     }
+    case CtrlMsg::DataBind: {
+      auto ptr = reinterpret_cast<const MobileGL::Wire::DataBind *>(obj);
+      return verifier.VerifyTable(ptr);
+    }
     default: return true;
   }
 }

MobileGL/MG_Remote/Protocol/mg_protocol_base.h

@@ -51,6 +51,14 @@ extern "C" {
 #define MOBILEGL_PROTOCOL_ABI_MAJOR 1
 #define MOBILEGL_PROTOCOL_ABI_MINOR 0
 
+// The CONTROL schema's revision (protocol.fbs), mixed into wireFingerprint. The fingerprint's
+// other inputs are struct sizes and codec digests of the DATA plane; nothing moved it when a
+// control message changed shape, so a peer built before a control-schema change agreed with one
+// built after it and the difference surfaced as whatever the first misread field did. Bump this
+// with every protocol.fbs change a peer must agree on.
+//   1  P7 wave 2-F, PH-7 (4): Welcome.dataNonce and the DataBind message.
+#define MOBILEGL_PROTOCOL_CONTROL_REVISION 1
+
 #define MOBILEGL_ABI_VERSION(major, minor) (((uint32_t)(major) << 16) | (uint32_t)(minor))
 #define MOBILEGL_ABI_MAJOR_OF(version) ((uint32_t)(version) >> 16)
 #define MOBILEGL_ABI_MINOR_OF(version) ((uint32_t)(version) & 0xFFFFu)

MobileGL/MG_Remote/Protocol/protocol.fbs

@@ -126,6 +126,20 @@ table Welcome {
   wireFingerprint: ulong;
   linkTerms: LinkTerms;
   backendType: uint;
+  // PH-7 (4), ID-P7-3. 128 bits from the server's CSPRNG, minted after the Hello authenticated
+  // and only on a Stream data plane. The client opens its data connection AFTER this Welcome
+  // and presents the value as that connection's first frame (DataBind); the server binds the
+  // connection to this session by the value, not by which connection arrived second. Absent on
+  // SharedSegments, whose descriptors travel over the control connection's own aux socket.
+  dataNonce: [ubyte];
+}
+
+// PH-7 (4). The FIRST and only control-framed message on a TCP data connection: the Welcome's
+// dataNonce, handed back. Everything after it on that connection is StreamLink's own framing.
+// A connection whose first frame is not this, or whose nonce is not the live session's, is
+// refused by name (`Refuse{Authentication}`) and never becomes a data plane.
+table DataBind {
+  nonce: [ubyte];
 }
 
 // ---------------------------------------------------------------------------
@@ -348,6 +362,7 @@ union CtrlMsg {
   LogLine,
   Refuse,
   LogFlush,
+  DataBind,
 }
 
 table CtrlEnvelope {

MobileGL/MG_Remote/Server/ServerMain.cpp

@@ -1,6 +1,7 @@
 // MobileGL - separate-process server and TCP session supervisor.
 // SPDX-License-Identifier: LGPL-3.0-only
 #include "../Transport/Doorbell.h"
+#include "../Transport/FdPassing.h"
 #include "../Transport/SocketTransport.h"
 #include "../Transport/WireLog.h"
 #include "../Protocol/SurfaceOpCodec.h"
@@ -14,6 +15,7 @@
 #include <MG_Backend/ServerRole.h>
 #include <MG_Util/Debug/Log.h>
 #include <MG_Util/Metrics/PipeStats.h>
+#include <algorithm>
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
@@ -21,6 +23,8 @@
 #include <string>
 #include <vector>
 #include <chrono>
+#include <cerrno>
+#include <poll.h>
 #include <sys/socket.h>
 #include <sys/wait.h>
 #include <unistd.h>
@@ -59,6 +63,103 @@ void RefuseBusy(SocketTransport& transport, const char* detail) {
     Refuse(transport, Protocol::RefuseCode::Busy, detail);
 }
 
+// PH-7 (4), ID-P7-3. THE DATA CONNECTION IS ROUTED BY WHAT IT SAYS, NOT BY WHEN IT CAME.
+//
+// On TCP the supervisor accepts ONE connection at a time. While no session is live, a
+// connection is a session's control connection and goes to a fresh child. While one is, the
+// connection's first frame decides: a DataBind is the live session's data connection and its
+// descriptor goes to that child over the hand-off socketpair, with the nonce it presented as
+// the sideband; anything else is a second client and is refused Busy exactly as before. The
+// supervisor never judges the nonce - ServerSession::BindDataConnection compares it, once, in
+// the child that minted it. This replaces AcceptPair's rule that the second connection to
+// arrive inside the window IS the data plane: over a Windows `adb forward` the two arrived
+// reordered, the data connection was read as control, and every child exited 67.
+constexpr std::uint32_t kFirstFrameWaitMs = 2000;
+constexpr std::uint64_t kFirstFrameMaxBytes = 1024 * 1024;
+
+void RouteWhileBusy(SocketTransport& connection, int handoff) {
+    std::vector<std::uint8_t> frame;
+    const int fd = connection.StreamFd();
+    const auto read = SocketTransport::ReceiveOneFrame(fd, kFirstFrameWaitMs, kFirstFrameMaxBytes, &frame);
+    std::uint8_t nonce[kDataNonceBytes] = {};
+    if (read == MOBILEGL_OK && DecodeDataBind(frame, nonce)) {
+        if (handoff >= 0 && FdPassing::SendFd(handoff, fd, MobileGLByteSpan{nonce, sizeof(nonce)}) == MOBILEGL_OK) {
+            // The child holds its own descriptor for this connection now; ours goes without a
+            // shutdown(2), which would disconnect the child's too.
+            connection.CloseLocalCopy();
+            return;
+        }
+        Refuse(connection, Protocol::RefuseCode::Authentication,
+               "data connection could not be handed to the live session");
+        return;
+    }
+    // The frame (if any) is consumed, so closing cannot RST the refusal away - the reason
+    // RefuseBusy reads the Hello first.
+    Refuse(connection, Protocol::RefuseCode::Busy, "one session is already active");
+}
+
+// True when the control peer has closed (EOF or error); false while it is open, readable or not.
+bool ControlPeerGone(int controlFd) {
+    char byte = 0;
+    const auto peeked = ::recv(controlFd, &byte, 1, MSG_PEEK | MSG_DONTWAIT);
+    return peeked == 0 || (peeked < 0 && errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR);
+}
+
+// The session child's source: descriptors the supervisor forwarded, each with the nonce its
+// DataBind presented. Watching the control connection too, so a client that went away after
+// Welcome ends the wait at once instead of holding the supervisor Busy for the whole deadline.
+Server::ServerSession::DataConnectionSource HandoffSource(int handoff, int controlFd) {
+    return [handoff, controlFd](std::uint32_t timeoutMs, int* outFd, std::uint8_t* outNonce) -> MobileGLResult {
+        pollfd fds[2] = {{handoff, POLLIN, 0}, {controlFd, POLLIN, 0}};
+        // A control connection with bytes waiting is not a hang-up and must not spin this loop:
+        // after the first look it is only asked again once per slice.
+        const int rc = ::poll(fds, 2, static_cast<int>(std::min<std::uint32_t>(timeoutMs, 50)));
+        if (rc < 0) return errno == EINTR ? MOBILEGL_ERR_TIMEOUT : MOBILEGL_ERR_TRANSPORT_CLOSED;
+        if ((fds[1].revents & (POLLIN | POLLHUP | POLLERR)) != 0 && ControlPeerGone(controlFd))
+            return MOBILEGL_ERR_TRANSPORT_CLOSED;
+        if ((fds[0].revents & (POLLIN | POLLHUP | POLLERR)) == 0) {
+            if ((fds[1].revents & POLLIN) != 0) {
+                pollfd only{handoff, POLLIN, 0};
+                if (::poll(&only, 1, static_cast<int>(std::min<std::uint32_t>(timeoutMs, 50))) <= 0)
+                    return MOBILEGL_ERR_TIMEOUT;
+            } else {
+                return MOBILEGL_ERR_TIMEOUT;
+            }
+        }
+        std::uint8_t sideband[FdPassing::kMaxSidebandBytes] = {};
+        std::uint64_t size = 0;
+        const auto received = FdPassing::ReceiveFd(handoff, outFd, MobileGLMutableByteSpan{sideband, sizeof(sideband)},
+                                                   &size, 0);
+        if (received != MOBILEGL_OK) return received;
+        // The supervisor always sends exactly the presented nonce; a short sideband compares as
+        // zeros, which no minted nonce is going to equal by accident.
+        std::memcpy(outNonce, sideband, kDataNonceBytes);
+        return MOBILEGL_OK;
+    };
+}
+
+// The single-session (no --serve) source: this process still owns the listener, so it accepts
+// the data connection itself and reads its DataBind the same way the supervisor would.
+Server::ServerSession::DataConnectionSource ListenerSource(int listener, int controlFd) {
+    return [listener, controlFd](std::uint32_t timeoutMs, int* outFd, std::uint8_t* outNonce) -> MobileGLResult {
+        if (ControlPeerGone(controlFd)) return MOBILEGL_ERR_TRANSPORT_CLOSED;
+        int fd = -1;
+        const auto accepted = SocketTransport::AcceptOne(listener, std::min<std::uint32_t>(timeoutMs, 250), &fd);
+        if (accepted != MOBILEGL_OK) return accepted;
+        std::vector<std::uint8_t> frame;
+        std::uint8_t presented[kDataNonceBytes] = {};
+        const auto read = SocketTransport::ReceiveOneFrame(fd, kFirstFrameWaitMs, kFirstFrameMaxBytes, &frame);
+        if (read != MOBILEGL_OK || !DecodeDataBind(frame, presented)) {
+            SocketTransport stray(fd, -1, TransportRole::Server);
+            Refuse(stray, Protocol::RefuseCode::Authentication, "data connection's first frame is not a DataBind");
+            return MOBILEGL_ERR_TIMEOUT;
+        }
+        std::memcpy(outNonce, presented, kDataNonceBytes);
+        *outFd = fd;
+        return MOBILEGL_OK;
+    };
+}
+
 struct FlushAck { ITransport* transport; std::uint64_t seq; };
 void SendLogAck(void* pointer) {
     auto& ack = *static_cast<FlushAck*>(pointer);
@@ -70,7 +171,10 @@ void SendLogAck(void* pointer) {
 }
 
 // Called only after fork: the supervisor never creates a backend or EGL context.
-[[noreturn]] void RunSession(std::unique_ptr<SocketTransport> control) {
+// `dataSource` is set on TCP (PH-7 (4)) and empty on a unix endpoint, whose second connection
+// is the SCM_RIGHTS socket AcceptPair already paired.
+[[noreturn]] void RunSession(std::unique_ptr<SocketTransport> control,
+                             Server::ServerSession::DataConnectionSource dataSource) {
     const int selfPid = static_cast<int>(::getpid());
     const bool tcp = control->IsTcp();
     std::vector<std::uint8_t> helloFrame;
@@ -81,6 +185,15 @@ void SendLogAck(void* pointer) {
     if (control->ReceiveFrame({helloFrame.data(), helloFrame.size()}, &bytes, 0) != MOBILEGL_OK) ::_exit(67);
     flatbuffers::Verifier verifier(helloFrame.data(), helloFrame.size());
     if (!Protocol::VerifyCtrlEnvelopeBuffer(verifier)) ::_exit(67);
+    // PH-7 (4). A DataBind with no live session to join: the supervisor routes a data connection
+    // to a child only while one is live, so this is a stale connection (its session already
+    // ended) or one that never had a session. Refused by name, and the child is a clean exit -
+    // it is a refusal, not a fault.
+    if (Protocol::GetCtrlEnvelope(helloFrame.data())->msg_type() == Protocol::CtrlMsg::DataBind) {
+        Refuse(*control, Protocol::RefuseCode::Authentication, "data connection names no live session");
+        std::fflush(nullptr);
+        ::_exit(0);
+    }
     const auto* hello = Protocol::GetCtrlEnvelope(helloFrame.data())->msg_as_Hello();
     if (!hello) ::_exit(67);
     // Reject incompatible wire before backend bring-up can obscure the cause.
@@ -122,11 +235,14 @@ void SendLogAck(void* pointer) {
         peerBell = std::make_unique<SocketDoorbell>(-1, clientBell[1], 1, true);
         session.SetExternalDoorbells(selfBell.get(), peerBell.get());
     }
+    if (tcp) session.SetDataConnectionSource(std::move(dataSource));
     const auto accepted = session.Accept(*control, &helloFrame);
     if (accepted != MOBILEGL_OK) {
         WireLogError("MG_Remote server: pid=%d Accept failed (rc=%d)", selfPid, static_cast<int>(accepted));
         std::fflush(nullptr);
-        ::_exit(accepted == MOBILEGL_ERR_PROTOCOL_MISMATCH ? 0 : 67);
+        // A refusal and a peer that went away before its data connection bound (PH-7 (4): the
+        // client closed after Welcome) are outcomes, not faults; the supervisor counts only 67.
+        ::_exit(accepted == MOBILEGL_ERR_PROTOCOL_MISMATCH || accepted == MOBILEGL_ERR_TRANSPORT_CLOSED ? 0 : 67);
     }
     if (!tcp)
     {
@@ -358,34 +474,74 @@ extern "C" __attribute__((visibility("default"))) int mobilegl_server_main(int a
     WireLogError("MG_Remote server: pid=%d listening on %s", static_cast<int>(::getpid()), endpoint.c_str());
     pid_t active = -1;
     unsigned long sessionsFaulted = 0;
+    const bool tcpEndpoint = endpoint.compare(0, 6, "tcp://") == 0;
+    // The supervisor's end of the live child's data-connection hand-off (PH-7 (4)); -1 while no
+    // session is live. It goes with the child: a reaped session can be handed nothing.
+    int handoff = -1;
+    const auto reap = [&]() {
+        const bool reaped = ReapActiveSession(active, sessionsFaulted);
+        if (active <= 0 && handoff >= 0) { ::close(handoff); handoff = -1; }
+        return reaped;
+    };
     for (;;) {
-        ReapActiveSession(active, sessionsFaulted);
+        reap();
         std::unique_ptr<SocketTransport> control;
-        const auto accepted = SocketTransport::AcceptPair(listener, serve ? 250 : 30000, control);
+        MobileGLResult accepted = MOBILEGL_ERR_UNSUPPORTED;
+        if (tcpEndpoint) {
+            int fd = -1;
+            accepted = SocketTransport::AcceptOne(listener, serve ? 250 : 30000, &fd);
+            if (accepted == MOBILEGL_OK) control = std::make_unique<SocketTransport>(fd, -1, TransportRole::Server);
+        } else {
+            accepted = SocketTransport::AcceptPair(listener, serve ? 250 : 30000, control);
+        }
         if (accepted == MOBILEGL_ERR_TIMEOUT && serve) continue;
         if (accepted != MOBILEGL_OK) { LogSupervisorSummary(sessionsFaulted); ::close(listener); return 73; }
         if (!serve) {
+            if (tcpEndpoint) {
+                // The listener stays open: this session's data connection has yet to arrive on it.
+                const int controlFd = control->StreamFd();
+                RunSession(std::move(control), ListenerSource(listener, controlFd));
+            }
             ::close(listener);
-            if (endpoint[0] != '@' && endpoint.compare(0, 6, "tcp://") != 0) ::unlink(endpoint.c_str());
-            RunSession(std::move(control));
+            if (endpoint[0] != '@') ::unlink(endpoint.c_str());
+            RunSession(std::move(control), {});
         }
-        ReapActiveSession(active, sessionsFaulted);
+        reap();
         // exit_group closes files just before waitpid can observe the exit.
         // Allow that small scheduling window; a live peer remains Busy.
         const auto reapDeadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(50);
         while (active > 0 && std::chrono::steady_clock::now() < reapDeadline) {
-            if (ReapActiveSession(active, sessionsFaulted)) break;
+            if (reap()) break;
             ::usleep(1000);
         }
         if (active > 0) {
-            RefuseBusy(*control, "one session is already active");
+            if (tcpEndpoint) RouteWhileBusy(*control, handoff);
+            else RefuseBusy(*control, "one session is already active");
+            continue;
+        }
+        int pair[2] = {-1, -1};
+        if (tcpEndpoint && FdPassing::CreateSocketPair(pair) != MOBILEGL_OK) {
+            RefuseBusy(*control, "could not create the data-connection hand-off");
             continue;
         }
         std::fflush(nullptr);
         const pid_t child = ::fork();
-        if (child == 0) { ::close(listener); RunSession(std::move(control)); }
-        if (child < 0) { RefuseBusy(*control, "could not create session child"); continue; }
+        if (child == 0) {
+            ::close(listener);
+            if (pair[0] >= 0) ::close(pair[0]);
+            const int controlFd = control->StreamFd();
+            RunSession(std::move(control), tcpEndpoint ? HandoffSource(pair[1], controlFd)
+                                                       : Server::ServerSession::DataConnectionSource{});
+        }
+        if (child < 0) {
+            if (pair[0] >= 0) ::close(pair[0]);
+            if (pair[1] >= 0) ::close(pair[1]);
+            RefuseBusy(*control, "could not create session child");
+            continue;
+        }
         active = child;
+        if (pair[1] >= 0) ::close(pair[1]);
+        handoff = pair[0];
         control->CloseLocalCopy();
     }
     LogSupervisorSummary(sessionsFaulted);

MobileGL/MG_Remote/Server/ServerSession.cpp

@@ -9,6 +9,7 @@
 // P5 package s1: the server half of a session.
 
 #include "../Transport/SocketTransport.h"
+#include "../Transport/StreamLink.h"
 #include "ServerSession.h"
 #include <MG_Remote/FatalFunnel.h>
 
@@ -558,9 +559,26 @@ namespace MobileGL::MG_Remote::Server {
 
         // ---- 3. the four segments, both control pages, the rings.
         MobileGLResult created = MOBILEGL_ERR_UNSUPPORTED;
+        // PH-7 (4), ID-P7-3. Minted HERE - after the token was checked, before a byte of Welcome
+        // exists - and bound after Welcome by BindDataConnection. The link is attached without
+        // its descriptor: Welcome has to announce the sizes of memory the connection it names
+        // will fill.
+        Uint8 dataNonce[Transport::kDataNonceBytes] = {};
+        Transport::StreamLink* pendingStream = nullptr;
         if (stream) {
             auto* socket = dynamic_cast<Transport::SocketTransport*>(&transport);
-            if (socket && socket->IsTcp()) created = AttachStreamLink(socket->TakeDataFd(), m_sizes);
+            if (socket == nullptr || !socket->IsTcp() || !m_dataSource)
+                return RefuseHandshake(transport, ::MobileGL::Wire::RefuseCode::LinkTerms,
+                                       "stream needs a TCP control connection and a data-connection source");
+            created = Transport::SocketTransport::MintNonce(dataNonce, sizeof(dataNonce));
+            if (created == MOBILEGL_OK) {
+                auto link = std::make_unique<Transport::StreamLink>();
+                created = link->AttachOwnedDeferred(m_sizes, Transport::TransportRoleTag::ServerConsumer);
+                if (created == MOBILEGL_OK) {
+                    pendingStream = link.get();
+                    AttachDataLink(std::move(link));
+                }
+            }
         } else {
             created = m_link->Memory().Create(m_sizes, Transport::MemoryRole::Server);
         }
@@ -597,6 +615,9 @@ namespace MobileGL::MG_Remote::Server {
             auto reply = segmentRef(3, ::MobileGL::Wire::SegmentKind::Reply, Slot::Reply);
             auto event = segmentRef(4, ::MobileGL::Wire::SegmentKind::Event, Slot::Event);
             auto stamp = builder.CreateString(BuildFingerprint());
+            const auto nonce = pendingStream != nullptr
+                ? builder.CreateVector(dataNonce, sizeof(dataNonce))
+                : ::flatbuffers::Offset<::flatbuffers::Vector<Uint8>>();
             const Uint64 maxReply = m_sizes.ReplyBytes / m_sizes.ReplySlotCount - sizeof(Transport::ReplySlotHeader);
             auto negotiated = ::MobileGL::Wire::CreateLinkTerms(builder, terms->dataPlane(),
                 ::MobileGL::Wire::WireForm::StructImage, maxReply, m_link->Memory().CmdRingCapacity(),
@@ -612,7 +633,8 @@ namespace MobileGL::MG_Remote::Server {
                 // equal by construction.
                 static_cast<Uint32>(SelfProcessId()),
                 cmd, stage, reply, event, stamp, ourFingerprint, ourFingerprint, negotiated,
-                m_backend == nullptr ? hello->backendType() : static_cast<Uint32>(m_backend->GetBackendType()));
+                m_backend == nullptr ? hello->backendType() : static_cast<Uint32>(m_backend->GetBackendType()),
+                nonce);
             auto root = ::MobileGL::Wire::CreateCtrlEnvelope(
                 builder, ::MobileGL::Wire::CtrlMsg::Welcome, welcome.Union());
             ::MobileGL::Wire::FinishCtrlEnvelopeBuffer(builder, root);
@@ -622,6 +644,13 @@ namespace MobileGL::MG_Remote::Server {
                 return sent;
             }
         }
+        if (pendingStream != nullptr) {
+            const MobileGLResult bound = BindDataConnection(transport, *pendingStream, dataNonce);
+            if (bound != MOBILEGL_OK) {
+                Close();
+                return bound;
+            }
+        }
 
         // ---- 5. the segment table and the ONE process-wide resolver.
         //
@@ -732,6 +761,52 @@ namespace MobileGL::MG_Remote::Server {
         return SendEnvelope(*m_transport, builder);
     }
 
+    // PH-7 (4), ID-P7-3. THE ONE PLACE A DATA CONNECTION IS BOUND TO A SESSION.
+    //
+    // Every connection the source offers is compared against this session's nonce, in constant
+    // time. A match becomes the Stream data plane. A mismatch is refused BY NAME - to the log,
+    // and to the peer on that same connection - and the wait goes on, so a stale connection
+    // left over from an earlier session, or one raced in by somebody who can reach the port,
+    // cannot end the session it failed to join. What ends the wait is the deadline or the
+    // control peer going away; the deadline is refused on the control connection, where the
+    // client is listening.
+    MobileGLResult ServerSession::BindDataConnection(Transport::ITransport& control,
+                                                     Transport::StreamLink& link, const Uint8* nonce) {
+        const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(kHandshakeTimeoutMs);
+        Uint32 refused = 0;
+        for (;;) {
+            const auto left = std::chrono::duration_cast<std::chrono::milliseconds>(
+                deadline - std::chrono::steady_clock::now()).count();
+            if (left <= 0) {
+                MGLOG_E("MG_Remote server: no data connection presented this session's nonce within "
+                        "%u ms (%u refused)", kHandshakeTimeoutMs, refused);
+                return RefuseHandshake(control, ::MobileGL::Wire::RefuseCode::Authentication,
+                                       "no data connection presented the session nonce");
+            }
+            int fd = -1;
+            Uint8 presented[Transport::kDataNonceBytes] = {};
+            const MobileGLResult offered = m_dataSource(static_cast<std::uint32_t>(left), &fd, presented);
+            if (offered == MOBILEGL_ERR_TIMEOUT) continue;
+            if (offered != MOBILEGL_OK) {
+                MGLOG_E("MG_Remote server: waiting for the data connection ended (rc=%d, %u refused)",
+                        static_cast<int>(offered), refused);
+                return offered;
+            }
+            if (Transport::ConstantTimeNonceMatch(nonce, presented)) {
+                const MobileGLResult bound = link.BindDataFd(fd);
+#if !defined(_WIN32)
+                if (bound != MOBILEGL_OK) ::close(fd);
+#endif
+                return bound;
+            }
+            ++refused;
+            // The stray's transport owns `fd` and closes it on the way out of this scope.
+            Transport::SocketTransport stray(fd, -1, Transport::TransportRole::Server);
+            (void)RefuseHandshake(stray, ::MobileGL::Wire::RefuseCode::Authentication,
+                                  "data connection nonce mismatch");
+        }
+    }
+
     MobileGLResult ServerSession::AttachStreamLink(int fd, const Transport::SessionSegmentSizes& sizes) {
         std::unique_ptr<Transport::ILink> link;
         const auto attached = Transport::CreateStreamLink(fd, sizes, Transport::TransportRoleTag::ServerConsumer, link);

MobileGL/MG_Remote/Server/ServerSession.h

@@ -42,6 +42,8 @@
 #pragma once
 #include "../Transport/ILink.h"
 #include <Includes.h>
+#include <cstdint>
+#include <functional>
 #include <vector>
 
 #include <MG_Pipe/MGPipe.h>
@@ -56,6 +58,10 @@
 #include "../Wire/PipeWireCodec.h"
 #include "PipeApplier.h"
 
+namespace MobileGL::MG_Remote::Transport {
+    class StreamLink;
+}
+
 namespace MobileGL::MG_Remote::Server {
 
     class ServerSession {
@@ -76,6 +82,20 @@ namespace MobileGL::MG_Remote::Server {
         MobileGLResult Accept(Transport::ITransport& transport,
                               const std::vector<Uint8>* firstFrame = nullptr);
 
+        // PH-7 (4), ID-P7-3. WHERE A TCP SESSION'S DATA CONNECTION COMES FROM.
+        //
+        // A Stream data plane is not handed to Accept with the control connection any more: the
+        // client opens it after Welcome and names this session by presenting Welcome.dataNonce
+        // as its first frame (DataBind). Whoever owns the listener - ServerMain's supervisor,
+        // which forwards the descriptor over a socketpair, or the single-session loop, which
+        // accepts it directly - reads that frame and offers the connection here with the nonce
+        // it presented. Accept compares, binds the one that matches, and refuses every other by
+        // name. MOBILEGL_ERR_TIMEOUT from the source means "nothing yet"; anything else ends
+        // the wait (TRANSPORT_CLOSED when the control peer went away while it waited).
+        using DataConnectionSource =
+            std::function<MobileGLResult(std::uint32_t timeoutMs, int* outFd, std::uint8_t* outNonce)>;
+        void SetDataConnectionSource(DataConnectionSource source) { m_dataSource = std::move(source); }
+
         // Re-publishes the whole snapshot. R-12: a SECOND arrival IS the invalidation signal,
         // which is how DirectGLES - which has no OnCapsInvalidated producer - tells the client
         // its InitCapabilities re-ran, without any dev-shaped backend edit. It is also the
@@ -207,6 +227,10 @@ namespace MobileGL::MG_Remote::Server {
         void LogMemory(const char* phase) const;
 
     private:
+        MobileGLResult BindDataConnection(Transport::ITransport& control, Transport::StreamLink& link,
+                                          const Uint8* nonce);
+
+        DataConnectionSource m_dataSource;
         std::unique_ptr<Transport::ILink> m_link;
         // sm: non-owning; the spawn entry point owns the SocketDoorbells.
         Transport::Doorbell* m_externalConsumerBell = nullptr;

MobileGL/MG_Remote/Transport/AuthToken.h

@@ -87,4 +87,19 @@ namespace MobileGL::MG_Remote::Transport {
         return difference == 0;
     }
 
+    // PH-7 (4), ID-P7-3. The width of `Welcome.dataNonce`: 128 bits from the CSPRNG, the same
+    // floor as the token for the same reason.
+    inline constexpr std::size_t kDataNonceBytes = 16;
+
+    // The nonce comparison. Both operands are exactly kDataNonceBytes by the time this is asked
+    // (a presented nonce of any other length is refused before it gets here), so the loop has
+    // one length and no early exit.
+    inline bool ConstantTimeNonceMatch(const unsigned char* expected, const unsigned char* presented) {
+        if (expected == nullptr || presented == nullptr) return false;
+        unsigned difference = 0;
+        for (std::size_t index = 0; index < kDataNonceBytes; ++index)
+            difference |= static_cast<unsigned>(expected[index] ^ presented[index]);
+        return difference == 0;
+    }
+
 } // namespace MobileGL::MG_Remote::Transport

MobileGL/MG_Remote/Transport/Ring.cpp

@@ -696,6 +696,7 @@ namespace MobileGL::MG_Remote::Transport {
         mix(inputs.ProgramArtifactsCodecVersion);
         mix(inputs.ProgramArtifactsSchema);
         mix(inputs.OpCount);
+        mix(inputs.ControlSchemaRevision);
         mix(inputs.AbiVersion);
         mix(inputs.PointerBits);
         mix(inputs.LittleEndian);

MobileGL/MG_Remote/Transport/SessionRings.h

@@ -573,6 +573,8 @@ namespace MobileGL::MG_Remote::Transport {
         std::uint64_t ProgramArtifactsCodecVersion = 0;
         std::uint64_t ProgramArtifactsSchema = 0;
         std::uint64_t OpCount = 0;
+        // MOBILEGL_PROTOCOL_CONTROL_REVISION (mg_protocol_base.h): the control schema's shape.
+        std::uint64_t ControlSchemaRevision = 0;
         std::uint32_t AbiVersion = 0;
         std::uint32_t PointerBits = 0;
         std::uint32_t LittleEndian = 0;

MobileGL/MG_Remote/Transport/SocketTransport.cpp

@@ -34,6 +34,9 @@
 #    include <netdb.h>
 #    include <netinet/in.h>
 #    include <netinet/tcp.h>
+#    if defined(__linux__)
+#        include <sys/syscall.h>
+#    endif
 #else
 #    define MOBILEGL_SOCKET_TRANSPORT_POSIX 0
 #endif
@@ -437,6 +440,126 @@ namespace MobileGL::MG_Remote::Transport {
         return MOBILEGL_OK;
     }
 
+    MobileGLResult SocketTransport::ConnectControl(const std::string& path, std::uint32_t timeoutMs,
+                                                   std::unique_ptr<SocketTransport>& outClient) {
+        if (!TcpName(path)) return ConnectTo(path, timeoutMs, outClient);
+        outClient.reset();
+        int control = -1;
+        const MobileGLResult connected = ConnectOne(path, timeoutMs, &control);
+        if (connected != MOBILEGL_OK) return connected;
+        outClient = std::make_unique<SocketTransport>(control, -1, TransportRole::Client);
+        return MOBILEGL_OK;
+    }
+
+    MobileGLResult SocketTransport::ConnectDataConnection(const std::string& path, std::uint32_t timeoutMs,
+                                                          MobileGLByteSpan firstFrame, int* outFd) {
+        if (outFd == nullptr || !TcpName(path)) return MOBILEGL_ERR_INVALID_ARGUMENT;
+        *outFd = -1;
+        std::vector<std::uint8_t> framed;
+        const MobileGLResult appended = AppendFrame(framed, firstFrame.data, firstFrame.size);
+        if (appended != MOBILEGL_OK) return appended;
+        int fd = -1;
+        const MobileGLResult connected = ConnectOne(path, timeoutMs, &fd);
+        if (connected != MOBILEGL_OK) return connected;
+        std::size_t written = 0;
+        while (written < framed.size()) {
+            const ssize_t n = ::send(fd, framed.data() + written, framed.size() - written, MSG_NOSIGNAL);
+            if (n > 0) { written += static_cast<std::size_t>(n); continue; }
+            if (n < 0 && errno == EINTR) continue;
+            WireLogError("MG_Remote SocketTransport: the data connection's DataBind could not be "
+                         "sent (%zu of %zu bytes): %s", written, framed.size(), std::strerror(errno));
+            ::close(fd);
+            return MOBILEGL_ERR_TRANSPORT_CLOSED;
+        }
+        *outFd = fd;
+        return MOBILEGL_OK;
+    }
+
+    MobileGLResult SocketTransport::AcceptOne(int listenFd, std::uint32_t timeoutMs, int* outFd) {
+        if (listenFd < 0 || outFd == nullptr) return MOBILEGL_ERR_INVALID_ARGUMENT;
+        *outFd = -1;
+        for (;;) {
+            const int ready = WaitReadable(listenFd, timeoutMs);
+            if (ready == 0) return MOBILEGL_ERR_TIMEOUT;
+            if (ready < 0) return MOBILEGL_ERR_TRANSPORT_CLOSED;
+            const int fd = ::accept(listenFd, nullptr, nullptr);
+            if (fd < 0) {
+                // The same two non-failures AcceptPair re-polls on.
+                if (errno == EINTR || errno == ECONNABORTED) continue;
+                WireLogError("MG_Remote SocketTransport: accept failed: %s", std::strerror(errno));
+                return MOBILEGL_ERR_TRANSPORT_CLOSED;
+            }
+            ::fcntl(fd, F_SETFD, FD_CLOEXEC);
+            if (TcpSocket(fd) && !ConfigureTcp(fd)) {
+                ::close(fd);
+                continue;
+            }
+            *outFd = fd;
+            return MOBILEGL_OK;
+        }
+    }
+
+    MobileGLResult SocketTransport::ReceiveOneFrame(int fd, std::uint32_t timeoutMs, std::uint64_t maxBytes,
+                                                    std::vector<std::uint8_t>* outPayload) {
+        if (fd < 0 || outPayload == nullptr) return MOBILEGL_ERR_INVALID_ARGUMENT;
+        outPayload->clear();
+        const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeoutMs);
+        const auto readExactly = [&](std::uint8_t* into, std::size_t size) -> MobileGLResult {
+            while (size != 0) {
+                const auto left = std::chrono::duration_cast<std::chrono::milliseconds>(
+                    deadline - std::chrono::steady_clock::now()).count();
+                if (left <= 0) return MOBILEGL_ERR_TIMEOUT;
+                const int ready = WaitReadable(fd, static_cast<std::uint32_t>(left));
+                if (ready == 0) return MOBILEGL_ERR_TIMEOUT;
+                if (ready < 0) return MOBILEGL_ERR_TRANSPORT_CLOSED;
+                const ssize_t n = ::recv(fd, into, size, 0);
+                if (n < 0 && errno == EINTR) continue;
+                if (n <= 0) return MOBILEGL_ERR_TRANSPORT_CLOSED;
+                into += n;
+                size -= static_cast<std::size_t>(n);
+            }
+            return MOBILEGL_OK;
+        };
+        std::uint8_t header[kFrameHeaderSize];
+        const MobileGLResult headerRead = readExactly(header, sizeof(header));
+        if (headerRead != MOBILEGL_OK) return headerRead;
+        std::uint32_t magic = 0, length = 0;
+        std::memcpy(&magic, header, sizeof(magic));
+        std::memcpy(&length, header + 4, sizeof(length));
+        if (magic != kFrameMagic || length > maxBytes) return MOBILEGL_ERR_PROTOCOL_MISMATCH;
+        outPayload->resize(length);
+        const MobileGLResult payloadRead = readExactly(outPayload->data(), length);
+        if (payloadRead != MOBILEGL_OK) outPayload->clear();
+        return payloadRead;
+    }
+
+    MobileGLResult SocketTransport::MintNonce(std::uint8_t* out, std::size_t size) {
+        if (out == nullptr) return MOBILEGL_ERR_INVALID_ARGUMENT;
+        std::size_t filled = 0;
+#if defined(__linux__) && defined(SYS_getrandom)
+        while (filled < size) {
+            const long n = ::syscall(SYS_getrandom, out + filled, size - filled, 0);
+            if (n > 0) { filled += static_cast<std::size_t>(n); continue; }
+            if (n < 0 && errno == EINTR) continue;
+            break; // ENOSYS on an old kernel: the device file below is the same pool.
+        }
+#endif
+        if (filled < size) {
+            const int source = ::open("/dev/urandom", O_RDONLY | O_CLOEXEC);
+            while (source >= 0 && filled < size) {
+                const ssize_t n = ::read(source, out + filled, size - filled);
+                if (n > 0) { filled += static_cast<std::size_t>(n); continue; }
+                if (n < 0 && errno == EINTR) continue;
+                break;
+            }
+            if (source >= 0) ::close(source);
+        }
+        if (filled == size) return MOBILEGL_OK;
+        WireLogError("MG_Remote SocketTransport: the kernel CSPRNG gave %zu of %zu bytes; a data "
+                     "nonce is not minted from anything weaker", filled, size);
+        return MOBILEGL_ERR_UNSUPPORTED;
+    }
+
     MobileGLResult SocketTransport::CreatePair(std::unique_ptr<SocketTransport>& outClient,
                                                std::unique_ptr<SocketTransport>& outServer) {
         outClient.reset();

MobileGL/MG_Remote/Transport/SocketTransport.h

@@ -100,6 +100,39 @@ namespace MobileGL::MG_Remote::Transport {
         static MobileGLResult ConnectTo(const std::string& path, std::uint32_t timeoutMs,
                                         std::unique_ptr<SocketTransport>& outClient);
 
+        // ---- PH-7 (4), ID-P7-3: a TCP data connection is BOUND, not paired --------
+        //
+        // AcceptPair/ConnectTo pair a session's two connections by ARRIVAL ORDER inside a
+        // window: first is control, second is data. That is a protocol only while nothing sits
+        // between the two processes. Over a Windows `adb forward` the two arrive reordered, the
+        // data connection is read as control, and every session child exits 67 waiting for a
+        // Hello that went down the other socket - and anybody who can reach the port can race a
+        // connection into the gap and become somebody else's data plane. So on TCP the session
+        // is ONE connection until it is authenticated; the server mints `Welcome.dataNonce`,
+        // and the client then opens its data connection and presents the nonce as that
+        // connection's first frame (DataBind). The server matches the value, not the order.
+        // Unix endpoints keep AcceptPair: their second connection is the SCM_RIGHTS socket of a
+        // same-user rendezvous, not a data plane anybody else can reach.
+
+        // Connects the control connection only on `tcp://`; identical to ConnectTo otherwise.
+        static MobileGLResult ConnectControl(const std::string& path, std::uint32_t timeoutMs,
+                                             std::unique_ptr<SocketTransport>& outClient);
+        // Opens one more connection to `path` and writes `firstFrame` (a CtrlEnvelope carrying
+        // DataBind, framed here) as its first bytes. *outFd is the caller's to attach a
+        // StreamLink to.
+        static MobileGLResult ConnectDataConnection(const std::string& path, std::uint32_t timeoutMs,
+                                                    MobileGLByteSpan firstFrame, int* outFd);
+        // Accepts ONE connection (TCP options applied, CLOEXEC). MOBILEGL_ERR_TIMEOUT when none.
+        static MobileGLResult AcceptOne(int listenFd, std::uint32_t timeoutMs, int* outFd);
+        // Reads EXACTLY one framed message from a raw socket - the header, then the payload,
+        // and not one byte more. A DataBind is followed on the same connection by StreamLink's
+        // own frames, which a buffered reader would swallow. PROTOCOL_MISMATCH for a bad magic
+        // or a payload over `maxBytes`, TIMEOUT when the frame did not complete in time.
+        static MobileGLResult ReceiveOneFrame(int fd, std::uint32_t timeoutMs, std::uint64_t maxBytes,
+                                              std::vector<std::uint8_t>* outPayload);
+        // `size` bytes from the kernel CSPRNG. Never a PRNG fallback: an error is an error.
+        static MobileGLResult MintNonce(std::uint8_t* out, std::size_t size);
+
         // Adopts an already-connected stream fd - what `sm` uses on each side of
         // the fork, where fd 3 is the stream and fd 4 is the aux socket.
         // `auxFd` may be -1: the transport then answers MOBILEGL_ERR_UNSUPPORTED

MobileGL/MG_Remote/Transport/StreamLink.cpp

@@ -330,21 +330,53 @@ namespace MobileGL::MG_Remote::Transport {
         (void)role;
         return MOBILEGL_ERR_UNSUPPORTED;
 #else
-        if (fd < 0 || !mirrors.Valid() || !mirrors.ReplySlotCount() ||
+        if (fd < 0) return MOBILEGL_ERR_INVALID_ARGUMENT;
+        const auto prepared = Prepare(mirrors, role);
+        if (prepared != MOBILEGL_OK) return prepared;
+        return BindDataFd(fd);
+#endif
+    }
+    MobileGLResult StreamLink::Prepare(SessionSegments& mirrors, TransportRoleTag role) {
+        if (!mirrors.Valid() || !mirrors.ReplySlotCount() ||
             mirrors.ReplyBytes() / mirrors.ReplySlotCount() <= sizeof(ReplySlotHeader))
             return MOBILEGL_ERR_INVALID_ARGUMENT;
         Detach();
         m_impl.reset(new Impl);
         auto& x = *m_impl;
-        x.fd = fd;
         x.memory = &mirrors;
         x.role = role;
         x.records = {mirrors.CmdRingBase(), mirrors.CmdRingCapacity(), mirrors.CmdRingCapacity() - 1, 0};
         x.recordCursor = {mirrors.CmdRingBase(), mirrors.CmdRingCapacity(), mirrors.CmdRingCapacity() - 1, 0};
         x.events = {mirrors.EventRingBase(), mirrors.EventRingCapacity(), mirrors.EventRingCapacity() - 1, 0};
         x.eventCursor = {mirrors.EventRingBase(), mirrors.EventRingCapacity(), mirrors.EventRingCapacity() - 1, 0};
+        return MOBILEGL_OK;
+    }
+    MobileGLResult StreamLink::BindDataFd(int fd) {
+#if defined(_WIN32)
+        (void)fd;
+        return MOBILEGL_ERR_UNSUPPORTED;
+#else
+        auto& x = *m_impl;
+        // Once, onto a prepared link: a second descriptor would be a second reader on one ring.
+        if (fd < 0 || !x.memory || x.fd >= 0 || x.reader.joinable()) return MOBILEGL_ERR_INVALID_ARGUMENT;
+        x.fd = fd;
         x.reader = std::thread([&x] { x.Loop(); });
         return MOBILEGL_OK;
+#endif
+    }
+    MobileGLResult StreamLink::AttachOwnedDeferred(const SessionSegmentSizes& sizes, TransportRoleTag role) {
+#if defined(_WIN32)
+        (void)sizes;
+        (void)role;
+        return MOBILEGL_ERR_UNSUPPORTED;
+#else
+        auto owned = std::make_unique<ShmLink>();
+        const auto result = owned->Memory().CreatePrivate(
+            sizes, role == TransportRoleTag::ClientProducer ? MemoryRole::Client : MemoryRole::Server);
+        if (result != MOBILEGL_OK) return result;
+        const auto prepared = Prepare(owned->Memory(), role);
+        if (prepared == MOBILEGL_OK) m_impl->owned = std::move(owned);
+        return prepared;
 #endif
     }
     MobileGLResult StreamLink::AttachOwned(int fd, const SessionSegmentSizes& sizes, TransportRoleTag role) {

MobileGL/MG_Remote/Transport/StreamLink.h

@@ -13,6 +13,14 @@ namespace MobileGL::MG_Remote::Transport {
         ~StreamLink() override;
         MobileGLResult Attach(int dataFd, SessionSegments& mirrors, TransportRoleTag role);
         MobileGLResult AttachOwned(int dataFd, const struct SessionSegmentSizes& sizes, TransportRoleTag role);
+        // PH-7 (4), ID-P7-3. The SERVER's half of a nonce-bound data connection, in two steps.
+        // The owned memory has to exist before the connection does - Welcome announces its
+        // sizes, and Welcome is what carries the nonce the client's data connection presents -
+        // so the link is attached WITHOUT a descriptor and without its reader thread, and
+        // BindDataFd starts both once the connection has proved which session it belongs to.
+        // Until then every write answers TRANSPORT_CLOSED (there is no fd) rather than blocking.
+        MobileGLResult AttachOwnedDeferred(const struct SessionSegmentSizes& sizes, TransportRoleTag role);
+        MobileGLResult BindDataFd(int dataFd);
         SessionSegments& Memory() override;
         void InitializeEndpoints() override;
         RingProducer& CommandsOut() override;
@@ -44,6 +52,7 @@ namespace MobileGL::MG_Remote::Transport {
         TransportRoleTag Role() const override;
 
     private:
+        MobileGLResult Prepare(SessionSegments& mirrors, TransportRoleTag role);
         struct Impl;
         std::unique_ptr<Impl> m_impl;
     };

MobileGL/MG_Test/Wire/SessionHandshakeTest.cpp

@@ -130,6 +130,7 @@ TEST(SessionHandshakeTest, TheAbiFingerprintChangesWhenAnyOfItsInputsDoes) {
     EXPECT_EQ(inputs.ProgramArtifactsCodecVersion, MG_State::GLState::kProgramArtifactsCodecVersion);
     EXPECT_EQ(inputs.ProgramArtifactsSchema, MG_State::GLState::ProgramArtifactsSchemaFingerprint());
     EXPECT_EQ(inputs.OpCount, static_cast<Uint64>(MG_Pipe::MGPWireOp::kOpCount));
+    EXPECT_EQ(inputs.ControlSchemaRevision, static_cast<Uint64>(MOBILEGL_PROTOCOL_CONTROL_REVISION));
     EXPECT_EQ(inputs.PointerBits, sizeof(void*) * 8);
     const auto perturbed = [&](auto mutate) {
         auto copy = inputs;
@@ -148,6 +149,7 @@ TEST(SessionHandshakeTest, TheAbiFingerprintChangesWhenAnyOfItsInputsDoes) {
     EXPECT_NE(production, perturbed([](auto& i) { ++i.ProgramArtifactsCodecVersion; })) << "ProgramArtifactsCodecVersion";
     EXPECT_NE(production, perturbed([](auto& i) { ++i.ProgramArtifactsSchema; })) << "ProgramArtifactsSchema";
     EXPECT_NE(production, perturbed([](auto& i) { ++i.OpCount; })) << "OpCount";
+    EXPECT_NE(production, perturbed([](auto& i) { ++i.ControlSchemaRevision; })) << "ControlSchemaRevision";
     EXPECT_NE(production, perturbed([](auto& i) { ++i.AbiVersion; })) << "AbiVersion";
     EXPECT_NE(production, perturbed([](auto& i) { ++i.PointerBits; })) << "PointerBits";
     EXPECT_NE(production, perturbed([](auto& i) { ++i.LittleEndian; })) << "LittleEndian";

MobileGL/MG_Test/Wire/SocketTransportTest.cpp

@@ -35,6 +35,7 @@
 // `BufferTooSmallKeepsTheMessage` matters more here: the message it must keep
 // is one this side already reassembled out of bytes nobody else can re-deliver.
 
+#include <MG_Remote/Handshake.h> // PH-7 (4): EncodeDataBind / DecodeDataBind
 #include <MG_Remote/Protocol/generated/protocol_generated.h> // RefuseCode, for the word below
 #include <MG_Remote/Transport/AuthToken.h> // PH-7 (3): the minimum and the comparison
 #include <MG_Remote/Transport/Doorbell.h> // kWaitForever
@@ -181,6 +182,75 @@ TEST(SocketTransportTest, TheTokenComparisonAnswersLengthAndPrefixTheSameWay) {
     EXPECT_FALSE(ConstantTimeTokenMatch(nullptr, "anything", 8));
 }
 
+// PH-7 (4), ID-P7-3. THE DATA CONNECTION'S FIRST FRAME IS READ EXACTLY, AND ONLY A DataBind
+// OF THE MINTED WIDTH IS ONE.
+//
+// A client writes its DataBind and then, on the same socket, StreamLink's own frames - possibly
+// in the same segment. The server hands that descriptor to a StreamLink reader after reading
+// the DataBind, so a reader that buffered ahead would eat the first command bytes and the
+// session would die of "invalid data frame header" one frame later. ReceiveOneFrame reads the
+// header and then exactly the payload; the bytes behind it are still on the socket afterwards.
+// End to end, over a real loopback TCP listener, through the two helpers ServerMain and the
+// client use. The two-process half (reordered arrival, stale nonce, refused by name) is
+// TcpLane.SupervisorProtocolControls.
+TEST(SocketTransportTest, TheDataBindIsReadExactlyAndTheStreamBehindItIsLeftOnTheSocket) {
+    int reservation = ::socket(AF_INET, SOCK_STREAM, 0);
+    ASSERT_GE(reservation, 0);
+    sockaddr_in address{};
+    address.sin_family = AF_INET;
+    address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    ASSERT_EQ(::bind(reservation, reinterpret_cast<sockaddr*>(&address), sizeof(address)), 0);
+    socklen_t addressSize = sizeof(address);
+    ASSERT_EQ(::getsockname(reservation, reinterpret_cast<sockaddr*>(&address), &addressSize), 0);
+    const auto endpoint = std::string("tcp://127.0.0.1:") + std::to_string(ntohs(address.sin_port));
+    ::close(reservation);
+    int listener = -1;
+    ASSERT_EQ(SocketTransport::Listen(endpoint, &listener), MOBILEGL_OK);
+
+    std::uint8_t nonce[kDataNonceBytes];
+    ASSERT_EQ(SocketTransport::MintNonce(nonce, sizeof(nonce)), MOBILEGL_OK);
+    const auto bind = MobileGL::MG_Remote::EncodeDataBind(nonce, sizeof(nonce));
+    int client = -1;
+    ASSERT_EQ(SocketTransport::ConnectDataConnection(endpoint, 1000, MobileGLByteSpan{bind.data(), bind.size()},
+                                                     &client), MOBILEGL_OK);
+    const char tail[] = "MGLD-the-first-stream-bytes";
+    ASSERT_EQ(::send(client, tail, sizeof(tail), MSG_NOSIGNAL), static_cast<ssize_t>(sizeof(tail)));
+
+    int accepted = -1;
+    ASSERT_EQ(SocketTransport::AcceptOne(listener, 1000, &accepted), MOBILEGL_OK);
+    std::vector<std::uint8_t> frame;
+    ASSERT_EQ(SocketTransport::ReceiveOneFrame(accepted, 1000, 1024, &frame), MOBILEGL_OK);
+    EXPECT_EQ(frame, bind);
+    std::uint8_t presented[kDataNonceBytes] = {};
+    ASSERT_TRUE(MobileGL::MG_Remote::DecodeDataBind(frame, presented));
+    EXPECT_TRUE(ConstantTimeNonceMatch(nonce, presented));
+    char after[sizeof(tail)] = {};
+    EXPECT_EQ(::recv(accepted, after, sizeof(after), MSG_WAITALL), static_cast<ssize_t>(sizeof(tail)));
+    EXPECT_EQ(std::memcmp(after, tail, sizeof(tail)), 0);
+
+    // Only a DataBind of the minted width is one: a shorter nonce, and a frame that is not a
+    // DataBind at all, both decode as "no", which the callers refuse by name.
+    const auto shortBind = MobileGL::MG_Remote::EncodeDataBind(nonce, sizeof(nonce) - 1);
+    EXPECT_FALSE(MobileGL::MG_Remote::DecodeDataBind(shortBind, presented));
+    std::vector<std::uint8_t> notABind(bind.size(), 0xA5);
+    EXPECT_FALSE(MobileGL::MG_Remote::DecodeDataBind(notABind, presented));
+    // One byte of difference anywhere is a mismatch.
+    presented[kDataNonceBytes - 1] ^= 1;
+    EXPECT_FALSE(ConstantTimeNonceMatch(nonce, presented));
+
+    // A frame over the caller's bound is refused before its payload is read.
+    int second = -1;
+    ASSERT_EQ(SocketTransport::ConnectDataConnection(endpoint, 1000, MobileGLByteSpan{bind.data(), bind.size()},
+                                                     &second), MOBILEGL_OK);
+    int big = -1;
+    ASSERT_EQ(SocketTransport::AcceptOne(listener, 1000, &big), MOBILEGL_OK);
+    EXPECT_EQ(SocketTransport::ReceiveOneFrame(big, 1000, bind.size() - 1, &frame), MOBILEGL_ERR_PROTOCOL_MISMATCH);
+    // And a connection that says nothing times out rather than blocking.
+    EXPECT_EQ(SocketTransport::AcceptOne(listener, 50, &big), MOBILEGL_ERR_TIMEOUT);
+    for (const int fd : {client, accepted, second, big, listener})
+        if (fd >= 0) ::close(fd);
+}
+
 namespace {
 
     MobileGLByteSpan Span(const std::string& text) {

scripts/ci/tcp_supervisor_smoke.py

@@ -92,7 +92,8 @@ def receive(peer, schema):
     if kind == schema['CtrlMsg'].CtrlMsg.Welcome:
         welcome = schema['Welcome'].Welcome()
         welcome.Init(table.Bytes, table.Pos)
-        return {'welcome': welcome.ServerPid(), 'fingerprint': welcome.WireFingerprint()}
+        return {'welcome': welcome.ServerPid(), 'fingerprint': welcome.WireFingerprint(),
+                'nonce': bytes(welcome.DataNonce(j) for j in range(welcome.DataNonceLength())).hex()}
     raise RuntimeError(f'unexpected first control reply {kind}')
 
 
@@ -127,6 +128,30 @@ def hello(schema, flatbuffers, fingerprint=0, major=1, token=kToken, identifier=
     return struct.pack('<4sI', b'MGLF', len(payload)) + payload
 
 
+def data_bind(schema, flatbuffers, nonce):
+    """PH-7 (4): the framed DataBind a client writes as its data connection's first bytes."""
+    builder = flatbuffers.Builder(64)
+    vector = builder.CreateByteVector(nonce)
+    message = schema['DataBind']
+    message.Start(builder)
+    message.AddNonce(builder, vector)
+    body = message.End(builder)
+    envelope = schema['CtrlEnvelope']
+    envelope.Start(builder)
+    envelope.AddMsgType(builder, schema['CtrlMsg'].CtrlMsg.DataBind)
+    envelope.AddMsg(builder, body)
+    builder.Finish(envelope.End(builder), b'MGLC')
+    payload = bytes(builder.Output())
+    return struct.pack('<4sI', b'MGLF', len(payload)) + payload
+
+
+def open_data(port, schema, flatbuffers, nonce):
+    """A data connection that has presented `nonce`. Left open: the server owns what happens next."""
+    data = socket.create_connection(('127.0.0.1', port), timeout=3)
+    data.sendall(data_bind(schema, flatbuffers, nonce))
+    return data
+
+
 def supervisor_environment(log, token, same_build=False):
     """The env a supervisor is started with, minus everything ServerMain's scrub catches."""
     env = dict(os.environ, MOBILEGL_IPC_ROLE='server', MOBILEGL_IPC_DIAL='no',
@@ -200,35 +225,39 @@ def supervisor(server, log, same_build=False):
 
 
 @contextmanager
-def held_session(port, schema, message):
+def held_session(port, schema, flatbuffers, message):
     """A welcomed session whose sockets STAY OPEN, so its child is still alive to be killed.
 
-    `exchange` closes the pair before returning, which is what every other control in this file
-    wants and is exactly wrong for the reap case: a child that has already exited cannot be made
-    to fault. Busy is retried here for the same reason it is there - a previous child may still
-    be reaching _exit.
+    `exchange` closes the connection before returning, which is what every other control in this
+    file wants and is exactly wrong for the reap case: a child that has already exited cannot be
+    made to fault. Busy is retried here for the same reason it is there - a previous child may
+    still be reaching _exit.
+
+    PH-7 (4): the data connection is opened AFTER the Welcome and presents its nonce, which is
+    what gets the child out of the handshake and into its control loop.
     """
     control = None
     data = None
     reply = None
     for _ in range(30):
         control = socket.create_connection(('127.0.0.1', port), timeout=3)
-        data = socket.create_connection(('127.0.0.1', port), timeout=3)
         control.sendall(message)
         reply = receive(control, schema)
         if reply.get('code') != 7:
             break
         control.close()
-        data.close()
         control = None
         time.sleep(.05)
     if control is None:
         raise RuntimeError('supervisor remained Busy while a held session was wanted')
+    if reply.get('welcome', 0):
+        data = open_data(port, schema, flatbuffers, bytes.fromhex(reply['nonce']))
     try:
         yield control, reply
     finally:
         control.close()
-        data.close()
+        if data is not None:
+            data.close()
 
 
 def last_faulted(path):
@@ -254,6 +283,10 @@ def wait_for_log(path, needle, seconds=15):
 
 @contextmanager
 def pair(port):
+    """A session's CONTROL connection. The name is historical: until PH-7 (4) a TCP session was
+    two connections paired by arrival order, and this opened both. The data connection is now
+    opened after Welcome with the nonce it carries (open_data), so a control that stops at the
+    handshake - every one but held_session's - never opens one."""
     deadline = time.monotonic() + 5
     while True:
         try:
@@ -264,11 +297,7 @@ def pair(port):
                 raise
             time.sleep(.02)
     try:
-        data = socket.create_connection(('127.0.0.1', port), timeout=3)
-        try:
-            yield control
-        finally:
-            data.close()
+        yield control
     finally:
         control.close()
 
@@ -308,7 +337,7 @@ def main():
         sys.path[:0] = [directory, str(ROOT / '3rdparty/flatbuffers/python')]
         import flatbuffers
         schema = {name: importlib.import_module('MobileGL.Wire.' + name)
-                  for name in ('CtrlEnvelope', 'CtrlMsg', 'Hello', 'LinkTerms', 'Refuse', 'Welcome')}
+                  for name in ('CtrlEnvelope', 'CtrlMsg', 'DataBind', 'Hello', 'LinkTerms', 'Refuse', 'Welcome')}
         evidence = {}
 
         # ---- PH-7 (2)(3), ID-P7-3: THE TWO LISTEN REFUSALS -------------------------------
@@ -364,7 +393,7 @@ def main():
             # a frame of a DIFFERENT schema that happened to verify was read here as a
             # CtrlEnvelope and its union tag believed. This is the cheap half of Ph's fuzz arm 1;
             # the arm itself (five malformed shapes, its own ctest lane) is scheduled after P7.
-            with held_session(port, schema,
+            with held_session(port, schema, flatbuffers,
                               hello(schema, flatbuffers, fingerprint=fingerprint)) as (control, welcome):
                 assert welcome.get('welcome', 0), welcome
                 control.sendall(hello(schema, flatbuffers, fingerprint=fingerprint,
@@ -390,7 +419,7 @@ def main():
             # the status, so the log carried no line at all for this event and the assertion
             # below could not have been written.
             held = hello(schema, flatbuffers, fingerprint=fingerprint)
-            with held_session(port, schema, held) as (control, welcome):
+            with held_session(port, schema, flatbuffers, held) as (control, welcome):
                 faulted = welcome.get('welcome', 0)
                 assert faulted and faulted != os.getpid(), welcome
                 # The count is a supervisor LIFETIME figure and the controls above have already
@@ -418,6 +447,64 @@ def main():
             after = exchange(port, schema, hello(schema, flatbuffers, fingerprint=fingerprint))
             assert after.get('welcome', 0) and after['welcome'] != faulted, after
             evidence['connect_after_fault'] = after
+
+            # ---- PH-7 (4), ID-P7-3: THE DATA CONNECTION IS BOUND BY NONCE ---------------------
+            #
+            # (a) THE FIELD FAILURE. Over a Windows `adb forward` the two connections of one
+            #     session arrived reordered, the server's AcceptPair read the data connection as
+            #     control, and every child exited 67. Reproduced here without adb: the data
+            #     connection (a DataBind with a nonce no live session minted) is presented FIRST,
+            #     the control connection second. It is refused by name, as a clean exit, and the
+            #     control connection behind it is welcomed.
+            stray = open_data(port, schema, flatbuffers, os.urandom(16))
+            try:
+                refused = receive(stray, schema)
+            finally:
+                stray.close()
+            assert refused.get('code') == 4 and refused.get('detail') == 'data connection names no live session', refused
+            reordered = exchange(port, schema, hello(schema, flatbuffers, fingerprint=fingerprint))
+            assert reordered.get('welcome', 0), reordered
+            assert len(bytes.fromhex(reordered['nonce'])) == 16, reordered
+            evidence['data_connection_first'] = refused
+
+            # (b) A STALE NONCE while a session is live: a data connection from an earlier session
+            #     (or raced in by anybody who can reach the port) is refused by name ON THAT
+            #     CONNECTION, and the session it failed to join still binds its own and comes up.
+            held = hello(schema, flatbuffers, fingerprint=fingerprint)
+            for _ in range(30):
+                control = socket.create_connection(('127.0.0.1', port), timeout=3)
+                control.sendall(held)
+                welcome = receive(control, schema)
+                if welcome.get('code') != 7:
+                    break
+                control.close()
+                time.sleep(.05)
+            try:
+                assert welcome.get('welcome', 0) and len(bytes.fromhex(welcome['nonce'])) == 16, welcome
+                nonce = bytes.fromhex(welcome['nonce'])
+                stale = bytes(b ^ 0xFF for b in nonce)
+                wrong = open_data(port, schema, flatbuffers, stale)
+                try:
+                    mismatch = receive(wrong, schema)
+                finally:
+                    wrong.close()
+                assert mismatch.get('code') == 4 and mismatch.get('detail') == 'data connection nonce mismatch', mismatch
+                right = open_data(port, schema, flatbuffers, nonce)
+                try:
+                    ready = wait_for_log(serverLog, f'pid={welcome["welcome"]} transport=spawn role=server ready')
+                    assert ready is not None, (
+                        f'session pid={welcome["welcome"]} did not come up on its own data connection '
+                        f'after a stale one was refused; see {serverLog}')
+                finally:
+                    right.close()
+            finally:
+                control.close()
+            named = wait_for_log(serverLog, 'Refuse{Authentication} data connection nonce mismatch')
+            assert named is not None, f'the stale data connection was not refused by name in {serverLog}'
+            evidence['stale_nonce'] = {'refusal': mismatch, 'line': next(
+                l.strip() for l in named.splitlines() if 'data connection nonce mismatch' in l)}
+            assert process.poll() is None, 'a refused data connection killed the supervisor'
+
         with supervisor(args.server, args.out / 'same-build-supervisor.log', same_build=True) as (port, _, _log):
             strict = exchange(port, schema, hello(schema, flatbuffers, fingerprint=fingerprint))
             assert strict.get('code') == 3, strict