diff --git a/src/workerd/api/sockets.c++ b/src/workerd/api/sockets.c++
index d6ccd2d166f..9a859cb4833 100644
--- a/src/workerd/api/sockets.c++
+++ b/src/workerd/api/sockets.c++
@@ -7,6 +7,7 @@
 #include "global-scope.h"
 #include "streams/standard.h"
 #include "system-streams.h"
+#include "worker-rpc.h"
 
 #include <workerd/io/io-context.h>
 #include <workerd/io/worker-interface.h>
@@ -14,10 +15,91 @@
 #include <workerd/jsg/url.h>
 #include <workerd/util/autogate.h>
 
+#include <capnp/compat/byte-stream.h>
+
 namespace workerd::api {
 
 namespace {
 
+// A kj::AsyncIoStream that bridges the two raw kj half-streams recovered when a Socket is
+// deserialized from RPC: the readable side (an AsyncInputStream unwrapped from the pushed
+// ByteStream) and the writable side (an ExplicitEndOutputStream over the peer's ByteStream).
+// setupSocket() builds the JS ReadableStream/WritableStream on top of this, exactly as it does for
+// a live socket.
+class TransferredSocketStream final: public kj::AsyncIoStream {
+ public:
+  TransferredSocketStream(
+      kj::Own<kj::AsyncInputStream> input, kj::Own<capnp::ExplicitEndOutputStream> output)
+      : input(kj::mv(input)),
+        output(kj::mv(output)) {}
+
+  kj::Promise<size_t> tryRead(void* buffer, size_t minBytes, size_t maxBytes) override {
+    return input->tryRead(buffer, minBytes, maxBytes);
+  }
+
+  kj::Promise<void> write(kj::ArrayPtr<const kj::byte> buffer) override {
+    return KJ_REQUIRE_NONNULL(output, "write after shutdownWrite() on transferred socket")
+        ->write(buffer);
+  }
+
+  kj::Promise<void> write(kj::ArrayPtr<const kj::ArrayPtr<const kj::byte>> pieces) override {
+    return KJ_REQUIRE_NONNULL(output, "write after shutdownWrite() on transferred socket")
+        ->write(pieces);
+  }
+
+  kj::Promise<void> whenWriteDisconnected() override {
+    // NOTE: For an RPC-transferred stream this reliably fires only on hard RPC breakage (the
+    // underlying CapnpToKjStreamAdapter rejects). A clean peer-initiated close is instead surfaced
+    // via readable EOF, which setupSocket() turns into a close when allowHalfOpen is false.
+    KJ_IF_SOME(out, output) {
+      return out->whenWriteDisconnected();
+    }
+    // The write side was already shut down, so there is nothing left to disconnect.
+    return kj::NEVER_DONE;
+  }
+
+  void shutdownWrite() override {
+    // ExplicitEndOutputStream needs an async end() to signal a clean EOF over the RPC ByteStream,
+    // but shutdownWrite() is synchronous. Kick off end() as an IoContext task so the clean-EOF RPC
+    // completes; dropping without end() would look like an abort to the peer. If there's no active
+    // IoContext (e.g. during teardown), just drop, which the ByteStream treats as an abort.
+    KJ_IF_SOME(out, output) {
+      auto owned = kj::mv(out);
+      output = kj::none;
+      if (IoContext::hasCurrent()) {
+        IoContext::current().addTask(
+            owned->end().attach(kj::mv(owned)).catch_([](kj::Exception&&) {}));
+      }
+    }
+  }
+
+  void abortRead() override {
+    // We cannot meaningfully abort the underlying RPC input stream mid-flight; it is released when
+    // this stream is destroyed. (A no-op also avoids canceling an in-flight tryRead().)
+  }
+
+  void getsockopt(int level, int option, void* value, uint* length) override {
+    KJ_UNIMPLEMENTED("getsockopt not available for transferred sockets");
+  }
+
+  void setsockopt(int level, int option, const void* value, uint length) override {
+    KJ_UNIMPLEMENTED("setsockopt not available for transferred sockets");
+  }
+
+  void getsockname(struct sockaddr* addr, uint* length) override {
+    KJ_UNIMPLEMENTED("getsockname not available for transferred sockets");
+  }
+
+  void getpeername(struct sockaddr* addr, uint* length) override {
+    KJ_UNIMPLEMENTED("getpeername not available for transferred sockets");
+  }
+
+ private:
+  kj::Own<kj::AsyncInputStream> input;
+  // Nulled by shutdownWrite(), which moves the stream out to drive a clean async end().
+  kj::Maybe<kj::Own<capnp::ExplicitEndOutputStream>> output;
+};
+
 // This function performs some basic length and characters checks, it does not guarantee that
 // the specified host is a valid domain. It should only be used to reject malicious
 // hosts.
@@ -63,6 +145,33 @@ SecureTransportKind parseSecureTransport(SocketOptions& opts) {
   }
 }
 
+// Maps a SecureTransportKind to its wire representation for socket RPC transfer.
+rpc::JsValue::External::SecureTransport toRpcSecureTransport(SecureTransportKind kind) {
+  switch (kind) {
+    case SecureTransportKind::OFF:
+      return rpc::JsValue::External::SecureTransport::OFF;
+    case SecureTransportKind::STARTTLS:
+      return rpc::JsValue::External::SecureTransport::STARTTLS;
+    case SecureTransportKind::ON:
+      return rpc::JsValue::External::SecureTransport::ON;
+  }
+  KJ_UNREACHABLE;
+}
+
+// Maps the wire representation of a transferred socket's security transport back to
+// SecureTransportKind.
+SecureTransportKind fromRpcSecureTransport(rpc::JsValue::External::SecureTransport transport) {
+  switch (transport) {
+    case rpc::JsValue::External::SecureTransport::OFF:
+      return SecureTransportKind::OFF;
+    case rpc::JsValue::External::SecureTransport::STARTTLS:
+      return SecureTransportKind::STARTTLS;
+    case rpc::JsValue::External::SecureTransport::ON:
+      return SecureTransportKind::ON;
+  }
+  KJ_UNREACHABLE;
+}
+
 bool getAllowHalfOpen(jsg::Optional<SocketOptions>& opts) {
   KJ_IF_SOME(o, opts) {
     return o.allowHalfOpen;
@@ -79,6 +188,53 @@ kj::Maybe<uint64_t> getWritableHighWaterMark(jsg::Optional<SocketOptions>& opts)
   return kj::none;
 }
 
+// Awaits a write-disconnect on `connection` and reports it through `fulfiller`: fulfills false when
+// the peer disconnects, or rejects on error.
+kj::Promise<void> handleDisconnected(
+    kj::AsyncIoStream& connection, kj::PromiseFulfiller<bool>& fulfiller) {
+  try {
+    co_await connection.whenWriteDisconnected();
+    fulfiller.fulfill(false);
+  } catch (...) {
+    fulfiller.reject(kj::getCaughtExceptionAsKj());
+  }
+}
+
+struct DisconnectWatcher {
+  // Resolves false when the connection reports write-disconnect, or true if the watch task is
+  // canceled (i.e. the Socket is GC'd) before that happens. Feed this to
+  // Socket::wireClosedToDisconnect() to resolve the `closed` promise. Awaiting it requires a
+  // JS-executing context.
+  kj::Promise<bool> disconnected;
+  // The kj task that watches for disconnect. It must be stored in the Socket so that it is canceled
+  // before the connection stream is destroyed, as required by KJ.
+  kj::Promise<void> watchTask;
+};
+
+// Sets up disconnection detection for `connection`.
+//
+// Disconnection handling is annoyingly complicated: we can't just `awaitIo(whenWriteDisconnected())`
+// directly, because the Socket could be GC'd before `whenWriteDisconnected()` completes, destroying
+// the underlying `connection`. By KJ rules we must cancel that promise before destroying the
+// connection, but there's no way to cancel a promise passed to `awaitIo()`. So the Socket holds the
+// watch task (`watchTask`), which waits for `whenWriteDisconnected()` and fulfills `disconnected`
+// with false; if the task is canceled first, a deferred fallback fulfills it with true instead.
+//
+// This part is pure kj and touches no JS, so it is safe to call where JS execution is disallowed
+// (e.g. Socket::deserialize()). The JS-side wiring that resolves `closed` is done separately by
+// Socket::wireClosedToDisconnect().
+DisconnectWatcher watchForDisconnect(kj::AsyncIoStream& connection) {
+  auto paf = kj::newPromiseAndFulfiller<bool>();
+  auto& fulfiller = *paf.fulfiller;
+  auto deferredCancel = kj::defer([fulfiller = kj::mv(paf.fulfiller)]() mutable {
+    // The watch task was canceled without fulfilling the fulfiller (the Socket was GC'd); silently
+    // fulfill with true so wireClosedToDisconnect() knows not to resolve `closed`.
+    fulfiller->fulfill(true);
+  });
+  auto watchTask = handleDisconnected(connection, fulfiller).attach(kj::mv(deferredCancel));
+  return DisconnectWatcher{.disconnected = kj::mv(paf.promise), .watchTask = kj::mv(watchTask)};
+}
+
 }  // namespace
 
 // Forward declarations
@@ -96,68 +252,14 @@ jsg::Ref<Socket> setupSocket(jsg::Lock& js,
     kj::Maybe<jsg::PromiseResolverPair<SocketInfo>> maybeOpenedPrPair) {
   auto& ioContext = IoContext::current();
 
-  // Disconnection handling is annoyingly complicated:
-  //
-  // We can't just context.awaitIo(connection->whenWriteDisconnected()) directly, because the
-  // Socket could be GC'ed before `whenWriteDisconnected()` completes, causing the underlying
-  // `connection` to be destroyed. By KJ rules, we are required to cancel the promise returned by
-  // `whenWriteDisconnected()` before destroying `connection`. But there's no way to cancel a
-  // promise passed to `context.awaitIo()`. We have to hold the promise directly in `Socket`, so
-  // that we can cancel it on destruction. But we *do* want to create a JS promise that resolves
-  // on disconnect, which is what awaitIo() would give us.
-  //
-  // So, we have to chain through a promise/fulfiller pair. The `Socket` holds
-  // `watchForDisconnectTask`, which is a `kj::Promise<void>` representing a task that waits for
-  // `whenWriteDisconnected()` and then fulfills the fulfiller end of `disconnectedPaf` with
-  // `false`. If the task is canceled, we instead fulfill `disconnectedPaf` with `true`.
-  //
-  // We then use `context.awaitIo()` to await the promise end of `disconnectedPaf`, and this gives
-  // us our `closed` promise. Well, almost...
-  //
-  // There's another wrinkle: There are some circumstances where we want to resolve the `closed`
-  // promise directly from an API call. We'd rather this did not have to drop out of the isolate
-  // and enter it a gain. So, our `awaitIo()` actually awaits a task that listens for the
-  // disconnected promise and then resolves some other JS resolver, `closedResolver`.
-  auto disconnectedPaf = kj::newPromiseAndFulfiller<bool>();
-  auto& disconnectedFulfiller = *disconnectedPaf.fulfiller;
-  auto deferredCancelDisconnected =
-      kj::defer([fulfiller = kj::mv(disconnectedPaf.fulfiller)]() mutable {
-    // In case the `whenWriteDisconected()` listener task is canceled without fulfilling the
-    // fulfiller, we want to silently fulfill it. This will happen when the Socket is GC'ed.
-    fulfiller->fulfill(true);
-  });
-
-  static auto constexpr handleDisconnected =
-      [](kj::AsyncIoStream& connection,
-          kj::PromiseFulfiller<bool>& fulfiller) -> kj::Promise<void> {
-    try {
-      co_await connection.whenWriteDisconnected();
-      fulfiller.fulfill(false);
-    } catch (...) {
-      auto exception = kj::getCaughtExceptionAsKj();
-      fulfiller.reject(kj::mv(exception));
-    }
-  };
-
-  auto watchForDisconnectTask = handleDisconnected(*connection, disconnectedFulfiller)
-                                    .attach(kj::mv(deferredCancelDisconnected));
+  // Set up disconnection detection. watchForDisconnect() builds the kj-side watch task (stored in
+  // the Socket so it is canceled before the connection stream is destroyed); the JS-side wiring that
+  // resolves `closed` is attached by wireClosedToDisconnect() after the Socket is constructed.
+  auto [disconnected, watchForDisconnectTask] = watchForDisconnect(*connection);
 
   auto closedPrPair = js.newPromiseAndResolver<void>();
   closedPrPair.promise.markAsHandled(js);
 
-  ioContext.awaitIo(js, kj::mv(disconnectedPaf.promise))
-      .then(
-          js, [resolver = closedPrPair.resolver.addRef(js)](jsg::Lock& js, bool canceled) mutable {
-    // We want to silently ignore the canceled case, without ever resolving anything. Note that
-    // if the application actually fetches the `closed` promise, then the JSG glue will prevent
-    // the socket from being GC'ed until that promise resolves, so it won't be canceled.
-    if (!canceled) {
-      resolver.resolve(js);
-    }
-  }, [resolver = closedPrPair.resolver.addRef(js)](jsg::Lock& js, jsg::Value exception) mutable {
-    resolver.reject(js, exception.getHandle(js));
-  });
-
   kj::Rc<kj::AsyncIoStream> refcountedConnection(kj::mv(connection));
   // Initialize the readable/writable streams with the readable/writable sides of an AsyncIoStream.
   auto sysStreams = newSystemMultiStream(refcountedConnection.addRef(), ioContext);
@@ -180,9 +282,11 @@ jsg::Ref<Socket> setupSocket(jsg::Lock& js,
       kj::mv(watchForDisconnectTask), kj::mv(options), kj::mv(tlsStarter), secureTransport,
       kj::mv(domain), isDefaultFetchPort, kj::mv(openedPrPair));
 
+  result->wireClosedToDisconnect(js, kj::mv(disconnected));
   KJ_IF_SOME(p, eofPromise) {
     result->handleReadableEof(js, kj::mv(p));
   }
+  result->trackOpenedState(js);
   return result;
 }
 
@@ -270,7 +374,6 @@ jsg::Ref<Socket> connectImpl(jsg::Lock& js,
   auto httpClient = asHttpClient(kj::mv(client));
   auto request = httpClient->connect(addressStr, *headers, httpConnectSettings);
   request.connection = request.connection.attach(kj::mv(httpClient));
-
   auto result = setupSocket(js, kj::mv(request.connection), kj::mv(addressStr),
       kj::none /* localAddress */, kj::mv(options), kj::mv(tlsStarter), secureTransport,
       kj::mv(domain), isDefaultFetchPort, kj::none /* maybeOpenedPrPair */);
@@ -522,6 +625,31 @@ void Socket::handleProxyError(jsg::Lock& js, kj::Exception e) {
   writable->getController().abort(js, js.error(e.getDescription())).markAsHandled(js);
 }
 
+void Socket::trackOpenedState(jsg::Lock& js) {
+  // whenResolved() creates a fresh branch off the `opened` promise without consuming
+  // `openedPromiseCopy` (which `close()` relies on). The branch settles in lockstep with `opened`.
+  openedPromiseCopy.whenResolved(js)
+      .then(js, [this, self = JSG_THIS](jsg::Lock&) { openedState = OpenedState::OPENED; },
+          [this, self = JSG_THIS](jsg::Lock&, jsg::Value&&) {
+    openedState = OpenedState::FAILED;
+  }).markAsHandled(js);
+}
+
+void Socket::wireClosedToDisconnect(jsg::Lock& js, kj::Promise<bool> disconnected) {
+  auto& context = IoContext::current();
+  context.awaitIo(js, kj::mv(disconnected))
+      .then(js, [this, self = JSG_THIS](jsg::Lock& js, bool canceled) {
+    // Silently ignore the canceled case (the Socket was GC'd before disconnect) without resolving
+    // anything. If the app actually awaited `closed`, JSG keeps the Socket alive until that promise
+    // resolves, so we won't be canceled in that case.
+    if (!canceled) {
+      closedResolver.resolve(js);
+    }
+  }, [this, self = JSG_THIS](jsg::Lock& js, jsg::Value exception) {
+    closedResolver.reject(js, exception.getHandle(js));
+  }).markAsHandled(js);
+}
+
 void Socket::handleReadableEof(jsg::Lock& js, jsg::Promise<void> onEof) {
   KJ_ASSERT(!getAllowHalfOpen(options));
   // Listen for EOF on the ReadableStream.
@@ -561,6 +689,193 @@ jsg::Promise<void> Socket::maybeCloseWriteSide(jsg::Lock& js) {
       }));
 }
 
+void Socket::serialize(jsg::Lock& js, jsg::Serializer& serializer) {
+  auto& handler = JSG_REQUIRE_NONNULL(
+      serializer.getExternalHandler(), DOMDataCloneError, "Socket can only be serialized for RPC.");
+  auto externalHandler = dynamic_cast<RpcSerializerExternalHandler*>(&handler);
+  JSG_REQUIRE(
+      externalHandler != nullptr, DOMDataCloneError, "Socket can only be serialized for RPC.");
+
+  // serialize() is synchronous and cannot await `opened`, so we require the caller to have already
+  // done so. This guarantees the connection is established and the SocketInfo (remote/local address)
+  // written below is authoritative, and avoids transferring a socket whose connection failed.
+  switch (openedState) {
+    case OpenedState::PENDING:
+      JSG_FAIL_REQUIRE(DOMDataCloneError,
+          "A Socket can only be serialized after it has opened. Await `socket.opened` before "
+          "transferring the socket over RPC.");
+    case OpenedState::FAILED:
+      JSG_FAIL_REQUIRE(DOMDataCloneError, "A Socket whose connection failed cannot be serialized.");
+    case OpenedState::OPENED:
+      break;
+  }
+
+  // Serialize the socket metadata, referencing the stream externals
+  externalHandler->write(
+      [this, remoteAddr = kj::str(remoteAddress), localAddr = mapCopyString(localAddress),
+          transport = toRpcSecureTransport(secureTransport),
+          allowHalfOpen = getAllowHalfOpen(options)](
+          rpc::JsValue::External::Builder builder) mutable {
+    auto socket = builder.initSocket();
+    socket.setRemoteAddress(remoteAddr);
+    socket.setSecureTransport(transport);
+    socket.setIsDefaultFetchPort(isDefaultFetchPort);
+    socket.setAllowHalfOpen(allowHalfOpen);
+    KJ_IF_SOME(local, localAddr) {
+      socket.setLocalAddress(local);
+    }
+  });
+
+  // Serialize the readable and writable streams as separate externals
+  readable->serialize(js, serializer);
+
+  writable->serialize(js, serializer);
+}
+
+jsg::Ref<Socket> Socket::deserialize(
+    jsg::Lock& js, rpc::SerializationTag tag, jsg::Deserializer& deserializer) {
+  auto& handler = JSG_REQUIRE_NONNULL(deserializer.getExternalHandler(), DOMDataCloneError,
+      "Socket can only be deserialized from RPC.");
+  auto externalHandler = dynamic_cast<RpcDeserializerExternalHandler*>(&handler);
+  JSG_REQUIRE(
+      externalHandler != nullptr, DOMDataCloneError, "Socket can only be deserialized from RPC.");
+
+  auto& ioContext = IoContext::current();
+
+  // Read the externals in the same order Socket::serialize() wrote them: (1) socket metadata,
+  // (2) the readable stream, (3) the writable stream. The stream externals are consumed here
+  // directly (rather than via ReadableStream/WritableStream::deserialize) so that we recover the
+  // raw kj half-streams and can rebuild a real AsyncIoStream backing the Socket.
+
+  // (1) Socket metadata.
+  auto socketExternal = externalHandler->read();
+  JSG_REQUIRE(socketExternal.isSocket(), DOMDataCloneError,
+      "external table slot type doesn't match serialization tag");
+  auto socketData = socketExternal.getSocket();
+  auto remoteAddr = kj::str(socketData.getRemoteAddress());
+  SecureTransportKind secureTransport = fromRpcSecureTransport(socketData.getSecureTransport());
+  bool allowHalfOpen = socketData.getAllowHalfOpen();
+  // An empty localAddress means the origin socket had none (the common case for outbound sockets).
+  kj::Maybe<kj::String> localAddr;
+  {
+    auto localText = socketData.getLocalAddress();
+    if (localText.size() > 0) {
+      localAddr = kj::str(localText);
+    }
+  }
+
+  // (2) Readable side: recover the raw input stream from the pushed ByteStream (mirrors
+  // ReadableStream::deserialize).
+  auto readableExternal = externalHandler->read();
+  JSG_REQUIRE(readableExternal.isReadableStream(), DOMDataCloneError,
+      "external table slot type doesn't match serialization tag");
+  auto rs = readableExternal.getReadableStream();
+  // Socket streams are always identity-encoded; newSystemMultiStream() assumes IDENTITY.
+  KJ_REQUIRE(rs.getEncoding() == StreamEncoding::IDENTITY,
+      "transferred socket readable must use identity encoding");
+  kj::Own<kj::AsyncInputStream> input = ioContext.getExternalPusher()->unwrapStream(rs.getStream());
+
+  // (3) Writable side: recover the raw output stream from the peer's ByteStream (mirrors
+  // WritableStream::deserialize).
+  auto writableExternal = externalHandler->read();
+  JSG_REQUIRE(writableExternal.isWritableStream(), DOMDataCloneError,
+      "external table slot type doesn't match serialization tag");
+  auto ws = writableExternal.getWritableStream();
+  KJ_REQUIRE(ws.getEncoding() == StreamEncoding::IDENTITY,
+      "transferred socket writable must use identity encoding");
+  kj::Own<capnp::ExplicitEndOutputStream> output =
+      ioContext.getByteStreamFactory().capnpToKjExplicitEnd(ws.getByteStream());
+
+  // Combine the two half-streams into a real AsyncIoStream (used for the connection stream, e.g. by
+  // startTls/takeConnectionStream) and derive the readable/writable sources from it.
+  kj::Own<kj::AsyncIoStream> asyncIoStream =
+      kj::heap<TransferredSocketStream>(kj::mv(input), kj::mv(output));
+  kj::Rc<kj::AsyncIoStream> refcountedConnection(kj::mv(asyncIoStream));
+  auto sysStreams = newSystemMultiStream(refcountedConnection.addRef(), ioContext);
+  // The underlying streams are bound to this IoContext (they disconnect when the RPC session's
+  // JsRpcCustomEvent is canceled), so their pumpTo() must not be deferred past the context's
+  // lifetime.
+  sysStreams.readable = newNoDeferredProxyReadableStream(kj::mv(sysStreams.readable), ioContext);
+  auto readable = js.alloc<ReadableStream>(ioContext, kj::mv(sysStreams.readable));
+
+  // Preserve the origin socket's half-open semantics: when allowHalfOpen is false, the write side is
+  // auto-closed once the read side reaches EOF. onEof() must be captured before `readable` is moved
+  // into the Socket below; the resulting promise is handed to handleReadableEof() after allocation.
+  kj::Maybe<jsg::Promise<void>> eofPromise;
+  if (!allowHalfOpen) {
+    eofPromise = readable->onEof(js);
+  }
+
+  auto closedPrPair = js.newPromiseAndResolver<void>();
+  closedPrPair.promise.markAsHandled(js);
+  auto openedPrPair = js.newPromiseAndResolver<SocketInfo>();
+  openedPrPair.promise.markAsHandled(js);
+
+  // No opened-gate on the writable: the connection is already established (opened resolves below),
+  // and passing a gate is unnecessary here.
+  auto writable = js.alloc<WritableStream>(ioContext, kj::mv(sysStreams.writable),
+      ioContext.getMetrics().tryCreateWritableByteStreamObserver());
+
+  // The connection is already established (the origin socket's `opened` was awaited before it could
+  // be serialized), so resolve `opened` immediately with the transferred SocketInfo. resolve() only
+  // schedules the settlement (handlers run later as microtasks), so it is safe under the deserialize
+  // scope.
+  openedPrPair.resolver.resolve(js,
+      SocketInfo{
+        .remoteAddress = kj::str(remoteAddr),
+        .localAddress = mapCopyString(localAddr),
+      });
+
+  // Set up disconnection detection now. This part is pure kj and safe under the deserialize scope;
+  // the JS wiring that resolves `closed` (wireClosedToDisconnect()) is attached in the deferred
+  // microtask below, since it cannot run where JS execution is disallowed.
+  auto [disconnected, watchForDisconnectTask] = watchForDisconnect(*refcountedConnection);
+
+  // No real TLS starter exists for a transferred socket, so startTls is unsupported.
+  auto tlsStarter = kj::heap<kj::TlsStarterCallback>();
+
+  // default fetch port
+  auto isDefaultFetchPort = socketData.getIsDefaultFetchPort();
+
+  // Reconstruct the subset of SocketOptions that affects the transferred socket's runtime behavior.
+  // Only allowHalfOpen is carried across transfer; getAllowHalfOpen() and the handleReadableEof()
+  // asserts rely on this reflecting the origin socket's setting.
+  SocketOptions options{.allowHalfOpen = allowHalfOpen};
+
+  auto socket = js.alloc<Socket>(js, ioContext, kj::mv(refcountedConnection), kj::str(remoteAddr),
+      kj::mv(localAddr), kj::mv(readable), kj::mv(writable), kj::mv(closedPrPair),
+      kj::mv(watchForDisconnectTask), kj::mv(options), kj::mv(tlsStarter), secureTransport,
+      kj::none /* domain */, isDefaultFetchPort, kj::mv(openedPrPair));
+
+  // handleReadableEof() and wireClosedToDisconnect() both attach jsg `.then()` continuations, which
+  // invoke V8 and are therefore forbidden inside the deserialize scope (JS execution is disallowed
+  // here). Defer them to microtasks that run once readValue() returns and JS execution is permitted
+  // again. Wrapping the function and enqueuing the microtask do not themselves invoke JS, so both are
+  // safe here. The kj-side signals (onEof above and the disconnect watcher) were already set up, so
+  // no event can be missed while the microtasks are pending.
+  //
+  // `disconnected` is a kj promise reachable from the JS-heap functor, so it is wrapped in an IoOwn
+  // to be held safely.
+  auto disconnectedOwn = ioContext.addObject(kj::heap<kj::Promise<bool>>(kj::mv(disconnected)));
+  js.v8Isolate->EnqueueMicrotask(js.wrapSimpleFunction(js.v8Context(),
+      JSG_VISITABLE_LAMBDA((self = socket.addRef(), disconnected = kj::mv(disconnectedOwn)), (self),
+          (jsg::Lock & js, const v8::FunctionCallbackInfo<v8::Value>& args) mutable {
+            self.get()->wireClosedToDisconnect(js, kj::mv(*disconnected));
+          })));
+  KJ_IF_SOME(p, eofPromise) {
+    js.v8Isolate->EnqueueMicrotask(js.wrapSimpleFunction(js.v8Context(),
+        JSG_VISITABLE_LAMBDA((self = socket.addRef(), eof = kj::mv(p)), (self, eof),
+            (jsg::Lock & js, const v8::FunctionCallbackInfo<v8::Value>& args) mutable {
+              self.get()->handleReadableEof(js, kj::mv(eof));
+            })));
+  }
+
+  // `opened` was resolved synchronously above, so the transferred socket is immediately in the
+  // OPENED state and can itself be re-serialized for a further RPC hop.
+  socket.get()->openedState = OpenedState::OPENED;
+  return socket;
+}
+
 jsg::Ref<Socket> SocketsModule::connect(
     jsg::Lock& js, AnySocketAddress address, jsg::Optional<SocketOptions> options) {
   return connectImpl(js, kj::none, kj::mv(address), kj::mv(options));
diff --git a/src/workerd/api/sockets.h b/src/workerd/api/sockets.h
index c25bb8c13ab..5ff50309fb5 100644
--- a/src/workerd/api/sockets.h
+++ b/src/workerd/api/sockets.h
@@ -151,6 +151,22 @@ class Socket: public jsg::Object {
   void handleReadableEof(jsg::Lock& js, jsg::Promise<void> onEof);
   // Sets up relevant callbacks to handle the case when the readable stream reaches EOF.
 
+  // Resolves the `closed` promise when `disconnected` (from watchForDisconnect()) reports a
+  // disconnect. This attaches jsg `.then()` continuations, so it must run in a JS-executing context:
+  // directly in setupSocket(), or deferred to a microtask in deserialize() (which runs under a scope
+  // that forbids JS execution).
+  void wireClosedToDisconnect(jsg::Lock& js, kj::Promise<bool> disconnected);
+
+  // Observes the `opened` promise and records its settled state in `openedState`. Must be called
+  // after allocation (it uses JSG_THIS, which requires an initialized refcount). This lets
+  // serialize() reject transfers of sockets that haven't finished connecting.
+  void trackOpenedState(jsg::Lock& js);
+
+  // RPC serialization support
+  void serialize(jsg::Lock& js, jsg::Serializer& serializer);
+  static jsg::Ref<Socket> deserialize(
+      jsg::Lock& js, rpc::SerializationTag tag, jsg::Deserializer& deserializer);
+
   JSG_RESOURCE_TYPE(Socket) {
     JSG_READONLY_PROTOTYPE_PROPERTY(readable, getReadable);
     JSG_READONLY_PROTOTYPE_PROPERTY(writable, getWritable);
@@ -166,6 +182,8 @@ class Socket: public jsg::Object {
     });
   }
 
+  JSG_SERIALIZABLE(rpc::SerializationTag::SOCKET);
+
   void visitForMemoryInfo(jsg::MemoryTracker& tracker) const {
     tracker.trackFieldWithSize("connectionData", sizeof(IoOwn<ConnectionData>));
     tracker.trackField("readable", readable);
@@ -222,6 +240,12 @@ class Socket: public jsg::Object {
   // Used to keep track of a pending `close` operation on the socket.
   bool isClosing = false;
 
+  // Tracks the settled state of `openedPromise`. A Socket can only be serialized for RPC transfer
+  // once its connection has been established (OPENED); serializing while still PENDING or after a
+  // FAILED connection throws, since serialize() is synchronous and cannot await `opened`.
+  enum class OpenedState : uint8_t { PENDING, OPENED, FAILED };
+  OpenedState openedState = OpenedState::PENDING;
+
   kj::Promise<kj::Own<kj::AsyncIoStream>> processConnection();
   jsg::Promise<void> maybeCloseWriteSide(jsg::Lock& js);
   jsg::Promise<void> closeImplOld(jsg::Lock& js);
diff --git a/src/workerd/api/streams/readable.c++ b/src/workerd/api/streams/readable.c++
index 9490f02b9d1..c2380983e84 100644
--- a/src/workerd/api/streams/readable.c++
+++ b/src/workerd/api/streams/readable.c++
@@ -678,6 +678,11 @@ class NoDeferredProxyReadableStream final: public ReadableStreamSource {
 
 }  // namespace
 
+kj::Own<ReadableStreamSource> newNoDeferredProxyReadableStream(
+    kj::Own<ReadableStreamSource> inner, IoContext& context) {
+  return kj::heap<NoDeferredProxyReadableStream>(kj::mv(inner), context);
+}
+
 void ReadableStream::serialize(jsg::Lock& js, jsg::Serializer& serializer) {
   // Serialize by effectively creating a `JsRpcStub` around this object and serializing that.
   // Except we don't actually want to do _exactly_ that, because we do not want to actually create
@@ -751,8 +756,8 @@ jsg::Ref<ReadableStream> ReadableStream::deserialize(
 
   kj::Own<kj::AsyncInputStream> in = ioctx.getExternalPusher()->unwrapStream(rs.getStream());
 
-  return js.alloc<ReadableStream>(ioctx,
-      kj::heap<NoDeferredProxyReadableStream>(newSystemStream(kj::mv(in), encoding, ioctx), ioctx));
+  return js.alloc<ReadableStream>(
+      ioctx, newNoDeferredProxyReadableStream(newSystemStream(kj::mv(in), encoding, ioctx), ioctx));
 }
 
 kj::StringPtr ReaderImpl::jsgGetMemoryName() const {
diff --git a/src/workerd/api/streams/readable.h b/src/workerd/api/streams/readable.h
index 61fd2a69b5f..b9d86a80440 100644
--- a/src/workerd/api/streams/readable.h
+++ b/src/workerd/api/streams/readable.h
@@ -555,4 +555,11 @@ class CountQueuingStrategy: public jsg::Object {
   QueuingStrategyInit init;
 };
 
+// Wraps a ReadableStreamSource so that pumpTo() is never deferred past the IoContext's lifetime.
+// Needed for RPC/session-bound sources (e.g. a Socket transferred over RPC), whose backing stream
+// disconnects when the IoContext is destroyed (the JsRpcCustomEvent is canceled). See the
+// implementation in readable.c++ for details.
+kj::Own<ReadableStreamSource> newNoDeferredProxyReadableStream(
+    kj::Own<ReadableStreamSource> inner, IoContext& context);
+
 }  // namespace workerd::api
diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel
index a11facc4e8b..5c173bd92e2 100644
--- a/src/workerd/api/tests/BUILD.bazel
+++ b/src/workerd/api/tests/BUILD.bazel
@@ -851,6 +851,12 @@ wd_test(
     tags = ["resources:socket:1"],
 )
 
+wd_test(
+    src = "outbound-interceptor-socket-test.wd-test",
+    args = ["--experimental"],
+    data = ["outbound-interceptor-socket-test.js"],
+)
+
 wd_test(
     src = "http-test-ts.ts-wd-test",
     args = ["--experimental"],
diff --git a/src/workerd/api/tests/js-rpc-socket-test.wd-test b/src/workerd/api/tests/js-rpc-socket-test.wd-test
index 4c871606ed7..391a2d92436 100644
--- a/src/workerd/api/tests/js-rpc-socket-test.wd-test
+++ b/src/workerd/api/tests/js-rpc-socket-test.wd-test
@@ -32,6 +32,9 @@ const unitTests :Workerd.Config = (
           (name = "defaultExport", service = "default-loop"),
           (name = "twelve", json = "12"),
           (name = "GreeterFactory", service = "GreeterFactory-loop"),
+          # Only present in the socket (loopback) variant: gates the socket-transfer assertions in
+          # js-rpc-test.js, which require a real RPC boundary to keep the producer context alive.
+          (name = "socketTransfer", json = "true"),
         ],
 
         durableObjectNamespaces = [
diff --git a/src/workerd/api/tests/js-rpc-test.js b/src/workerd/api/tests/js-rpc-test.js
index e4c35eabfe2..1ea1357ebee 100644
--- a/src/workerd/api/tests/js-rpc-test.js
+++ b/src/workerd/api/tests/js-rpc-test.js
@@ -282,6 +282,25 @@ export class MyService extends WorkerEntrypoint {
     return func;
   }
 
+  async getSocket() {
+    // Connect to our own connect() handler (which writes "hello" and closes) and return the client
+    // socket. Returning it over RPC transfers the socket to the caller, which can then read the
+    // bytes back through the reconstructed (transferred) socket.
+    const socket = this.env.MyService.connect('localhost:1234');
+    // A Socket can only be serialized for RPC once its connection has been established; await
+    // `opened` before returning it (Socket::serialize() is synchronous and cannot await this).
+    await socket.opened;
+    return socket;
+  }
+
+  async getEchoSocket() {
+    // Connect to the default entrypoint's echo connect() handler and return the client socket, so
+    // the caller can transfer it over RPC and exercise a write→peer→read round-trip.
+    const socket = this.env.defaultExport.connect('localhost:1234');
+    await socket.opened;
+    return socket;
+  }
+
   getRpcPromise(callback) {
     return callback();
   }
@@ -607,6 +626,12 @@ export default class DefaultService extends WorkerEntrypoint {
     // Test this.env here just to prove omitting the constructor entirely works.
     return new Response('default service ' + this.env.twelve);
   }
+
+  // Echoes everything written to the socket back to the reader. Used by getEchoSocket() so the
+  // socket-transfer test can exercise a write→peer→read round-trip through a transferred socket.
+  async connect(socket) {
+    await socket.readable.pipeTo(socket.writable);
+  }
 }
 
 export let basicServiceBinding = {
@@ -878,6 +903,53 @@ export let namedServiceBinding = {
         'Could not serialize object of type "Object". This type does not support ' +
         'serialization.',
     });
+    // Socket transfer over RPC is only exercised in the socket (loopback) variant of this test.
+    // The in-process variant can't sustain it: once getSocket() returns, the producer's request
+    // context tears down (there is no persistent RPC session to hold it open), so the byte pump
+    // feeding the transferred readable stops. The `socketTransfer` marker binding is present only in
+    // js-rpc-socket-test.wd-test.
+    if (env.socketTransfer) {
+      // A Socket returned over RPC is transferred to the caller. Verify a real byte round-trip
+      // through the reconstructed (transferred) socket: getSocket() connects to our connect()
+      // handler which writes "hello" and closes, so reading to EOF must yield "hello".
+      {
+        const socket = await env.MyService.getSocket();
+
+        // The `opened` metadata must survive the transfer: getSocket() connected to
+        // 'localhost:1234', so the deserialized socket should report that as its remote address.
+        const info = await socket.opened;
+        assert.strictEqual(info.remoteAddress, 'localhost:1234');
+
+        const dec = new TextDecoder();
+        let result = '';
+        for await (const chunk of socket.readable) {
+          result += dec.decode(chunk, { stream: true });
+        }
+        result += dec.decode();
+        assert.strictEqual(result, 'hello');
+      }
+
+      // Exercise the write direction through a transferred socket: getEchoSocket() connects to an
+      // echo handler, so bytes we write must come back unchanged. This proves both socket.writable
+      // and socket.readable survive the RPC transfer.
+      {
+        const socket = await env.MyService.getEchoSocket();
+        await socket.opened;
+
+        const enc = new TextEncoder();
+        const dec = new TextDecoder();
+        const writer = socket.writable.getWriter();
+        await writer.write(enc.encode('ping'));
+        await writer.close();
+
+        let result = '';
+        for await (const chunk of socket.readable) {
+          result += dec.decode(chunk, { stream: true });
+        }
+        result += dec.decode();
+        assert.strictEqual(result, 'ping');
+      }
+    }
 
     // A stateless entryponit method that never returns should fail due to PendingEvent tracking.
     await assert.rejects(() => env.MyService.neverReturn(), {
diff --git a/src/workerd/api/tests/outbound-interceptor-socket-test.js b/src/workerd/api/tests/outbound-interceptor-socket-test.js
new file mode 100644
index 00000000000..09735d412b5
--- /dev/null
+++ b/src/workerd/api/tests/outbound-interceptor-socket-test.js
@@ -0,0 +1,167 @@
+// Copyright (c) 2026 Cloudflare, Inc.
+// Licensed under the Apache 2.0 license found in the LICENSE file or at:
+//     https://opensource.org/licenses/Apache-2.0
+
+// Generic "outbound interceptor" socket-transfer test.
+//
+// This mirrors a proxy/interceptor service that, rather than connecting to a backend itself, obtains
+// a socket from a downstream service over JS RPC and returns it to its own caller. The returned
+// Socket is therefore transferred across TWO RPC hops (backend connect() handler -> interceptor ->
+// client), exercising Socket::serialize()/deserialize() at each hop.
+import { WorkerEntrypoint } from 'cloudflare:workers';
+import assert from 'node:assert';
+
+// Downstream backend, reached via `env.BACKEND.connect()`. Echoes everything written back to the
+// reader so the client can prove both directions of the transferred socket survive.
+export class Backend extends WorkerEntrypoint {
+  async connect(socket) {
+    await socket.readable.pipeTo(socket.writable);
+  }
+}
+
+// Downstream backend that writes a marker and then half-closes: it closes only its write side (so
+// the client sees read EOF) while keeping its read side open by draining it. This lets the client
+// observe whether the transferred socket honors `allowHalfOpen` (i.e. whether the client auto-closes
+// its own write side once its read side hits EOF).
+export class HalfCloseBackend extends WorkerEntrypoint {
+  async connect(socket) {
+    const writer = socket.writable.getWriter();
+    await writer.write(new TextEncoder().encode('bye'));
+    await writer.close();
+    // Keep the read side open, draining anything the client sends until it closes its write side.
+    for await (const _chunk of socket.readable) {
+      // discard
+    }
+  }
+}
+
+// The interceptor. Mirrors the outbound-interceptor structure (a `host` check that routes "magic"
+// addresses to a specific backend), but returns the backend's socket over RPC rather than piping it
+// -- which is the pattern whose lifetime we want to validate.
+export class OutboundInterceptor extends WorkerEntrypoint {
+  async interceptConnect(host) {
+    // `this.env.BACKEND.connect(host)` does NOT call Backend's `connect(socket)` handler as an
+    // RPC method. `connect` is a reserved name on a service binding (Fetcher): this invokes the
+    // built-in client-side socket API `Fetcher::connect` (C++: `jsg::Ref<Socket> connect(...)` in
+    // api/http.{h,c++} -> connectImpl in api/sockets.c++), which RETURNS the *client* end of a new
+    // connection to the BACKEND service. workerd then dispatches the *server* end of that same
+    // connection to Backend's exported `connect(socket)` ingress handler (the echo above). So there
+    // are two ends of one pipe: what we return here is the client end.
+    const socket = this.env.BACKEND.connect(host);
+    // A Socket can only be serialized for RPC once its connection has been established, so we must
+    // await `opened` before returning it. Socket::serialize() is synchronous and cannot await this
+    // itself; transferring a still-connecting socket throws a DataCloneError.
+    await socket.opened;
+    return socket;
+  }
+
+  // Returns a socket WITHOUT awaiting `opened` first. Serializing it for the RPC response must throw
+  // a DataCloneError, since the connection state (and SocketInfo) isn't yet settled.
+  interceptConnectWithoutAwait(host) {
+    return this.env.BACKEND.connect(host);
+  }
+
+  // Connects to the half-close backend with the given `allowHalfOpen` option and transfers the
+  // socket to the caller. The option must survive serialization so the transferred socket behaves
+  // the same as a locally-created one.
+  async interceptConnectHalfOpen(host, allowHalfOpen) {
+    const socket = this.env.HALFCLOSE.connect(host, { allowHalfOpen });
+    await socket.opened;
+    return socket;
+  }
+}
+
+async function readToEnd(socket) {
+  const dec = new TextDecoder();
+  let result = '';
+  for await (const chunk of socket.readable) {
+    result += dec.decode(chunk, { stream: true });
+  }
+  result += dec.decode();
+  return result;
+}
+
+async function echoRoundTrip(socket, payload) {
+  await socket.opened;
+  const enc = new TextEncoder();
+  const dec = new TextDecoder();
+  const writer = socket.writable.getWriter();
+  await writer.write(enc.encode(payload));
+  await writer.close();
+  let result = '';
+  for await (const chunk of socket.readable) {
+    result += dec.decode(chunk, { stream: true });
+  }
+  result += dec.decode();
+  return result;
+}
+
+export let interceptorForwardsSocket = {
+  async test(ctrl, env) {
+    // interceptConnect() returns a socket obtained from the backend over RPC (backend -> interceptor);
+    // awaiting it here transfers it a second hop (interceptor -> client). If the producer context is
+    // not kept alive past interceptConnect()'s return, the echo round-trip below would hang.
+    const socket = await env.INTERCEPTOR.interceptConnect('backend:5432');
+    const echoed = await echoRoundTrip(socket, 'ping');
+    assert.strictEqual(echoed, 'ping');
+  },
+};
+
+export let transferringUnopenedSocketThrows = {
+  async test(ctrl, env) {
+    // A socket that hasn't finished connecting cannot be serialized for RPC. The interceptor returns
+    // one without awaiting `opened`, so serializing the RPC response must fail with a DataCloneError.
+    await assert.rejects(
+      env.INTERCEPTOR.interceptConnectWithoutAwait('backend:5432'),
+      {
+        name: 'DataCloneError',
+      }
+    );
+  },
+};
+
+export let transferredSocketHonorsAllowHalfOpenFalse = {
+  async test(ctrl, env) {
+    // With allowHalfOpen=false (the default), a transferred socket must auto-close its write side
+    // once the read side reaches EOF. The backend half-closes after sending "bye", so reading to EOF
+    // triggers the auto-close, which resolves `closed`.
+    const socket = await env.INTERCEPTOR.interceptConnectHalfOpen(
+      'halfclose:1',
+      false
+    );
+    assert.strictEqual(await readToEnd(socket), 'bye');
+    // If allowHalfOpen were not carried across transfer, the write side would stay open and `closed`
+    // would hang here.
+    await socket.closed;
+  },
+};
+
+export let transferredSocketHonorsAllowHalfOpenTrue = {
+  async test(ctrl, env) {
+    // With allowHalfOpen=true, the write side must stay open after the read side reaches EOF, so we
+    // can still write to the (still-reading) backend afterwards.
+    const socket = await env.INTERCEPTOR.interceptConnectHalfOpen(
+      'halfclose:1',
+      true
+    );
+    assert.strictEqual(await readToEnd(socket), 'bye');
+    // The write side must not have been auto-closed; writing after read EOF must succeed. If the
+    // transferred socket wrongly defaulted to allowHalfOpen=false, this write would throw.
+    const writer = socket.writable.getWriter();
+    await writer.write(new TextEncoder().encode('still-open'));
+    await writer.close();
+  },
+};
+
+export let transferredSocketClosedResolvesOnExplicitClose = {
+  async test(ctrl, env) {
+    // A transferred socket must still support explicit close(): calling it resolves `closed`.
+    // allowHalfOpen=true so nothing auto-closes the socket for us -- only the explicit close() does.
+    const socket = await env.INTERCEPTOR.interceptConnectHalfOpen(
+      'halfclose:1',
+      true
+    );
+    await socket.close();
+    await socket.closed;
+  },
+};
diff --git a/src/workerd/api/tests/outbound-interceptor-socket-test.wd-test b/src/workerd/api/tests/outbound-interceptor-socket-test.wd-test
new file mode 100644
index 00000000000..2b7905469b6
--- /dev/null
+++ b/src/workerd/api/tests/outbound-interceptor-socket-test.wd-test
@@ -0,0 +1,61 @@
+# Verifies that a Socket returned as a JS RPC value keeps working after the producing method
+# returns, including when forwarded through an intermediate "interceptor" service (two RPC hops).
+using Workerd = import "/workerd/workerd.capnp";
+
+const unitTests :Workerd.Config = (
+  services = [
+    ( name = "outbound-interceptor-socket-test",
+      worker = (
+        modules = [
+          (name = "worker", esModule = embed "outbound-interceptor-socket-test.js")
+        ],
+        compatibilityFlags = ["nodejs_compat", "experimental"],
+        bindings = [
+          # Reached by the test to obtain a forwarded socket.
+          (name = "INTERCEPTOR", service = "interceptor-loop"),
+          # Used by OutboundInterceptor to reach the downstream backend. Present on the shared worker
+          # definition so the OutboundInterceptor entrypoint sees it in `this.env`.
+          (name = "BACKEND", service = "backend-loop"),
+          # Half-close backend, used to verify allowHalfOpen survives socket transfer.
+          (name = "HALFCLOSE", service = "halfclose-loop"),
+        ],
+      )
+    ),
+    ( name = "interceptor-loop",
+      external = (
+        address = "loopback:interceptor-loop",
+        http = (capnpConnectHost = "cappy")
+      )
+    ),
+    ( name = "backend-loop",
+      external = (
+        address = "loopback:backend-loop",
+        http = (capnpConnectHost = "cappy")
+      )
+    ),
+    ( name = "halfclose-loop",
+      external = (
+        address = "loopback:halfclose-loop",
+        http = (capnpConnectHost = "cappy")
+      )
+    ),
+    ( name = "internet", network = ( allow = ["private"] ) ),
+  ],
+  sockets = [
+    ( name = "interceptor-loop",
+      address = "loopback:interceptor-loop",
+      service = (name = "outbound-interceptor-socket-test", entrypoint = "OutboundInterceptor"),
+      http = (capnpConnectHost = "cappy")
+    ),
+    ( name = "backend-loop",
+      address = "loopback:backend-loop",
+      service = (name = "outbound-interceptor-socket-test", entrypoint = "Backend"),
+      http = (capnpConnectHost = "cappy")
+    ),
+    ( name = "halfclose-loop",
+      address = "loopback:halfclose-loop",
+      service = (name = "outbound-interceptor-socket-test", entrypoint = "HalfCloseBackend"),
+      http = (capnpConnectHost = "cappy")
+    ),
+  ],
+);
diff --git a/src/workerd/io/worker-interface.capnp b/src/workerd/io/worker-interface.capnp
index 83164e2e4d8..6c72fee8bfa 100644
--- a/src/workerd/io/worker-interface.capnp
+++ b/src/workerd/io/worker-interface.capnp
@@ -493,6 +493,11 @@ enum SerializationTag {
 
   blob @15;
   # A Blob, serialized as its MIME type followed by its raw bytes.
+
+  socket @16;
+  # A transferred Socket. Serialized as socket metadata (see External.socket) followed by its
+  # readable and writable streams, which are emitted as separate readableStream/writableStream
+  # externals.
 }
 
 enum StreamEncoding {
@@ -524,6 +529,13 @@ struct JsValue {
   # (We could also call these "capabilities", but that word is pretty overloaded already.)
 
   struct External {
+    enum SecureTransport {
+      # Security transport mode for a transferred Socket. Mirrors api::SecureTransportKind.
+      off @0;
+      starttls @1;
+      on @2;
+    }
+
     union {
       invalid @0 :Void;
       # Invalid default value to reduce confusion if an External wasn't initialized properly.
@@ -591,6 +603,27 @@ struct JsValue {
       # guarantees), so instead we send it with a placeholder representing the token to be provided
       # later.
 
+      socket :group {
+        # A Socket. Like ReadableStream, this requires bi-directional stream communication.
+        remoteAddress @16 :Text;
+        # The remote address this socket is connected to.
+
+        secureTransport @17 :SecureTransport;
+        # Security transport mode.
+
+        isDefaultFetchPort @18 :Bool;
+        # Indicates whether the socket is connected to port 443 or 80.
+
+        localAddress @19 :Text;
+        # The local address of the socket, if known (e.g. the CONNECT authority for inbound
+        # sockets). Empty for outbound sockets where no useful local address exists.
+
+        allowHalfOpen @20 :Bool;
+        # Whether the socket allows the read and write sides to close independently. When false, the
+        # runtime auto-closes the write side once the read side reaches EOF, so this must be carried
+        # across transfer to preserve the origin socket's half-open semantics.
+      }
+
       # TODO(soon): WebSocket, Request, Response
     }
   }
diff --git a/tools/cross/format.py b/tools/cross/format.py
index 6ef2291b368..26b1fe39d54 100755
--- a/tools/cross/format.py
+++ b/tools/cross/format.py
@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+from __future__ import annotations
 
 import json
 import logging