Split Session into login, routing, transfer and play packages

Session delegates frontend login to FrontendLogin, first-backend dialing to
InitialRouter and the whole transfer state machine to TransferCoordinator
behind the TransferHost port. Per-frame PLAY handlers move to session.play and
dialing helpers to session.channel. Relay hot path is unchanged.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

CLAUDE.md

@@ -32,6 +32,8 @@ The slow tests are the session integration tests (`ProxySessionListenerTest`, `S
 
 **Session threading.** `ProxySessionListener` owns the listener, event loops and the UUID→session index. Each `Session` (one per player connection) confines its mutable state to the frontend event loop, and its backend channels are created on that same loop (`SessionChannels.backendBootstrap` enforces it). Login verification, backend DNS and plugin callbacks run off that loop and re-enter it by posting a task; results are checked against the current connection identity so a stale async result cannot act on a reconnected player. Plugin work goes through bounded pools that reject when full.
 
+**Session layout.** `core.session` holds `Session` (the Netty handler and lifecycle owner) plus the pieces it delegates to: `FrontendLogin` (handshake, status, encryption, Mojang verification), `InitialRouter` (placement decision and first backend dial), `BackendHandshakes`. `core.session.transfer` is `TransferCoordinator`, which reaches the session only through the `TransferHost` port (Session supplies it as an adapter) and owns all transfer state. `core.session.play` has the per-frame PLAY handlers and relay start-up buffering; `core.session.channel` has backend dialing and pipeline lookups. Dependencies point one way: `session` -> `transfer` -> `play`/`channel`.
+
 **Data path.** After login, `RawRelay` forwards already-framed PLAY frames between the two channels without copying. Per-frame concerns are separate inbound handlers (`KeepAliveBridge`, `PlayerCommandInterceptor`, `TabCompletionBridge`, `TransferFrameHandler`) in front of the relay. Backpressure works by pausing reads on the source when the peer is not writable; the relay flushes once per read batch (`channelReadComplete`).
 
 **Transfers.** `Session.beginTransfer` dials and logs into the candidate backend while the old link keeps serving, buffers frames from both sides (`TransferFrameBuffer`, `TransitionBuffer`), swaps relays, and rewrites the player's entity IDs afterwards (`Minecraft1710EntityIds`). Forge clients also need an FML reset/handshake before the world switch; `TransferCandidate` and `PlayObservation` track that state. Failure before the client-visible switch restores the old link; after it, the session is closed.

benchmarks/RelayBenchmark.java

@@ -1,5 +1,7 @@
 package dev.moonbridge.core.session;
 
+import dev.moonbridge.core.session.play.KeepAliveBridge;
+
 import dev.moonbridge.core.protocol.MinecraftFrameDecoder;
 import dev.moonbridge.core.protocol.ProtocolProfile;
 import dev.moonbridge.core.protocol.ProtocolVarInt;

docs/architecture.md

@@ -28,7 +28,7 @@ flowchart LR
     F --> G[PLAY / Forge 握手与转发]
 ```
 
-普通转发沿用协议 5 帧边界，不重新编码普通帧；同一次 socket 读取解码出的帧逐个写入对端，在读取完成时合并为一次 flush，读取之外转发的帧由每批一次的事件循环任务补发 flush。前后端通道绑定同一 Netty EventLoop，会话控制状态在该循环串行变更。Keep Alive ID 由代理转换并跟踪未完成请求；必要的玩家实体 ID 数据包在转服后改写。通道不可写时通过暂停对端读取施加背压。实现见 [`Session`](../proxy-core/src/main/java/dev/moonbridge/core/session/Session.java)、[`RawRelay`](../proxy-core/src/main/java/dev/moonbridge/core/relay/RawRelay.java) 与 [`KeepAliveBridge`](../proxy-core/src/main/java/dev/moonbridge/core/session/KeepAliveBridge.java)。
+普通转发沿用协议 5 帧边界，不重新编码普通帧；同一次 socket 读取解码出的帧逐个写入对端，在读取完成时合并为一次 flush，读取之外转发的帧由每批一次的事件循环任务补发 flush。前后端通道绑定同一 Netty EventLoop，会话控制状态在该循环串行变更。Keep Alive ID 由代理转换并跟踪未完成请求；必要的玩家实体 ID 数据包在转服后改写。通道不可写时通过暂停对端读取施加背压。实现见 [`Session`](../proxy-core/src/main/java/dev/moonbridge/core/session/Session.java)、[`RawRelay`](../proxy-core/src/main/java/dev/moonbridge/core/relay/RawRelay.java) 与 [`KeepAliveBridge`](../proxy-core/src/main/java/dev/moonbridge/core/session/play/KeepAliveBridge.java)。
 
 ## 后端目录与发现所有权
 
@@ -62,6 +62,6 @@ flowchart LR
 ## 相关源码
 
 - 插件 API：[`PluginContext`](../proxy-plugin-api/src/main/java/dev/moonbridge/api/PluginContext.java)、[`Servers`](../proxy-plugin-api/src/main/java/dev/moonbridge/api/Servers.java)、[`Players`](../proxy-plugin-api/src/main/java/dev/moonbridge/api/Players.java)
-- 会话和转服：[`Session`](../proxy-core/src/main/java/dev/moonbridge/core/session/Session.java)、[`TransferCandidate`](../proxy-core/src/main/java/dev/moonbridge/core/session/TransferCandidate.java)、[`TransferFrameBuffer`](../proxy-core/src/main/java/dev/moonbridge/core/session/TransferFrameBuffer.java)
+- 会话和转服：[`Session`](../proxy-core/src/main/java/dev/moonbridge/core/session/Session.java)、[`TransferCandidate`](../proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferCandidate.java)、[`TransferFrameBuffer`](../proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferFrameBuffer.java)
 - 目录：[`BackendCatalog`](../proxy-core/src/main/java/dev/moonbridge/core/backend/BackendCatalog.java)、[`InMemoryBackendCatalog`](../proxy-core/src/main/java/dev/moonbridge/core/backend/InMemoryBackendCatalog.java)
 - 后端控制通道：[`BackendControlService`](../proxy-core/src/main/java/dev/moonbridge/core/control/BackendControlService.java)

docs/operations.md

@@ -71,7 +71,7 @@ Docker 容器间的监听、服务名和地址配置见[配置与部署](backend
 现有测试可用于定位行为范围：
 
 - 登录、认证和协议：`proxy-core/src/test/java/dev/moonbridge/core/auth/` 与 `proxy-core/src/test/java/dev/moonbridge/core/protocol/`
-- 会话、断开、转服与缓冲：`proxy-core/src/test/java/dev/moonbridge/core/session/`，重点包括 `SessionTransferTest`、`SessionTransferDispatchTest`、`TransferFrameBufferTest`
+- 会话、断开、转服与缓冲：`proxy-core/src/test/java/dev/moonbridge/core/session/`，重点包括 `SessionTransferTest`、`SessionTransferDispatchTest`、`transfer/TransferFrameBufferTest`
 - 插件生命周期、访问检查和 API：`proxy-core/src/test/java/dev/moonbridge/core/plugin/PluginHostTest.java`、`proxy-core/src/test/java/dev/moonbridge/core/session/ProxySessionListenerTest.java`、`proxy-core/src/test/java/dev/moonbridge/app/ProxyConfigurationTest.java` 与 `proxy-plugin-api/src/test/`
 - 控制通道：`proxy-core/src/test/java/dev/moonbridge/core/control/` 与 `backend-channel-client/src/test/`
 

proxy-core/src/main/java/dev/moonbridge/core/session/BackendHandshakes.java

@@ -0,0 +1,51 @@
+package dev.moonbridge.core.session;
+
+import dev.moonbridge.api.PlayerIdentity;
+import dev.moonbridge.core.auth.VerifiedProfile;
+import dev.moonbridge.core.forwarding.BungeeLegacyForwarding;
+import dev.moonbridge.core.protocol.MinecraftHandshake;
+import dev.moonbridge.core.protocol.ProtocolProfile;
+import dev.moonbridge.messaging.session.ForwardedSessionProof;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+
+/** Builds the handshake a backend receives: forwarded identity plus, when bound, a session proof. */
+final class BackendHandshakes {
+    private BackendHandshakes() { }
+
+    static ByteBuf encode(ProxySessionListener owner, Channel frontend, MinecraftHandshake handshake,
+                          PlayerIdentity identity, String username, VerifiedProfile verifiedProfile,
+                          String backendName, long backendEpoch) {
+        byte[] secret = owner.sessionBindingSecret(backendName);
+        try {
+            if (owner.onlineMode()) {
+                return forward(owner, frontend, handshake, identity, verifiedProfile, secret, backendName, backendEpoch);
+            }
+            if (secret == null) return handshake.encode(frontend.alloc(), ProtocolProfile.minecraft1710());
+            // OFFLINE identity is derived by this proxy and is not a Mojang-authenticated profile.
+            // Only forward it when this backend has explicitly configured session binding.
+            VerifiedProfile offlineProfile = new VerifiedProfile(identity.playerId(), username, List.of());
+            return forward(owner, frontend, handshake, identity, offlineProfile, secret, backendName, backendEpoch);
+        } finally {
+            if (secret != null) Arrays.fill(secret, (byte) 0);
+        }
+    }
+
+    private static ByteBuf forward(ProxySessionListener owner, Channel frontend, MinecraftHandshake handshake,
+                                   PlayerIdentity identity, VerifiedProfile profile, byte[] secret,
+                                   String backendName, long backendEpoch) {
+        String proof = null;
+        if (secret != null && backendEpoch > 0) {
+            long expiresAt = Math.addExact(System.currentTimeMillis(), ForwardedSessionProof.MAX_LIFETIME_MILLIS);
+            proof = ForwardedSessionProof.create(owner.proxyEpoch(), identity.playerId(), identity.connectionId(),
+                    backendName, backendEpoch, UUID.randomUUID(), expiresAt, secret);
+        }
+        return BungeeLegacyForwarding.encode(frontend.alloc(), handshake,
+                (InetSocketAddress) frontend.remoteAddress(), profile, proof);
+    }
+}

proxy-core/src/main/java/dev/moonbridge/core/session/FrontendLogin.java

@@ -0,0 +1,184 @@
+package dev.moonbridge.core.session;
+
+import dev.moonbridge.core.auth.AuthenticatedEncryption;
+import dev.moonbridge.core.auth.MinecraftCipherDecoder;
+import dev.moonbridge.core.auth.MinecraftCipherEncoder;
+import dev.moonbridge.core.auth.MinecraftEncryptionRequest;
+import dev.moonbridge.core.auth.MinecraftEncryptionResponse;
+import dev.moonbridge.core.auth.OnlineModeCrypto;
+import dev.moonbridge.core.auth.VerifiedProfile;
+import dev.moonbridge.core.protocol.LoginStart;
+import dev.moonbridge.core.protocol.MinecraftFrameDecoder;
+import dev.moonbridge.core.protocol.MinecraftHandshake;
+import dev.moonbridge.core.protocol.ProtocolProfile;
+import dev.moonbridge.core.protocol.ProtocolVarInt;
+import dev.moonbridge.core.session.play.FrameTransformHandler;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
+
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.security.GeneralSecurityException;
+import java.util.Arrays;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
+import java.util.function.BooleanSupplier;
+
+/**
+ * The client-facing half of connecting: handshake, server-list status, login start and (online mode)
+ * encryption and Mojang verification. It ends by reporting who the player is; everything after that
+ * (permissions, admission, placement) belongs to the session. Confined to the frontend event loop.
+ */
+final class FrontendLogin {
+    private static final String LOGIN_WAIT_GUARD = "login-wait-guard";
+
+    private final ProxySessionListener owner;
+    private final Channel frontend;
+    private final Runnable closeSession;
+    private final BooleanSupplier shuttingDown;
+    private final BiConsumer<UUID, String> authenticated;
+    private MinecraftHandshake handshake;
+    private boolean statusRequest;
+    private LoginStart loginStart;
+    private MinecraftEncryptionRequest encryptionRequest;
+    private VerifiedProfile verifiedProfile;
+    private CompletableFuture<Optional<VerifiedProfile>> verification;
+
+    FrontendLogin(ProxySessionListener owner, Channel frontend, Runnable closeSession,
+                  BooleanSupplier shuttingDown, BiConsumer<UUID, String> authenticated) {
+        this.owner = owner;
+        this.frontend = frontend;
+        this.closeSession = closeSession;
+        this.shuttingDown = shuttingDown;
+        this.authenticated = authenticated;
+    }
+
+    MinecraftHandshake handshake() { return handshake; }
+    LoginStart loginStart() { return loginStart; }
+    VerifiedProfile verifiedProfile() { return verifiedProfile; }
+
+    void cancel() {
+        if (verification != null) verification.cancel(true);
+    }
+
+    /** Any frame the client sends while the proxy is still deciding about the login is a protocol error. */
+    void guardPendingLogin() {
+        ChannelPipeline pipeline = frontend.pipeline();
+        if (pipeline.get(LOGIN_WAIT_GUARD) == null) {
+            pipeline.addFirst(LOGIN_WAIT_GUARD, new FrameTransformHandler(closeSession) {
+                @Override protected ByteBuf transform(ChannelHandlerContext ctx, ByteBuf frame) {
+                    throw new IllegalStateException("client sent data while login was pending");
+                }
+            });
+        }
+        frontend.config().setAutoRead(true);
+    }
+
+    void releaseGuard() {
+        if (frontend.pipeline().get(LOGIN_WAIT_GUARD) != null) frontend.pipeline().remove(LOGIN_WAIT_GUARD);
+    }
+
+    void receive(ByteBuf packet) {
+        if (handshake == null) {
+            receiveHandshake(packet);
+            return;
+        }
+        if (statusRequest) {
+            receiveStatus(packet);
+            return;
+        }
+        if (encryptionRequest != null) {
+            receiveEncryptionResponse(packet);
+            return;
+        }
+        if (loginStart != null) {
+            closeSession.run();
+            return;
+        }
+        loginStart = LoginStart.decode(packet, ProtocolProfile.minecraft1710());
+        if (owner.onlineMode()) {
+            encryptionRequest = owner.newEncryptionRequest();
+            frontend.writeAndFlush(encryptionRequest.encode(frontend.alloc())).addListener(write -> {
+                if (!write.isSuccess()) closeSession.run();
+            });
+            return;
+        }
+        UUID uuid = UUID.nameUUIDFromBytes(("OfflinePlayer:" + loginStart.username()).getBytes(StandardCharsets.UTF_8));
+        authenticated.accept(uuid, loginStart.username());
+    }
+
+    private void receiveHandshake(ByteBuf packet) {
+        handshake = MinecraftHandshake.decode(packet, ProtocolProfile.minecraft1710());
+        if (handshake.nextState() == MinecraftHandshake.NextState.LOGIN) {
+            int separator = handshake.serverAddress().indexOf('\0');
+            if (separator >= 0) {
+                handshake = new MinecraftHandshake(handshake.protocolVersion(),
+                        handshake.serverAddress().substring(0, separator), handshake.serverPort(), handshake.nextState());
+            }
+        }
+        if (handshake.protocolVersion() != ProtocolProfile.PROTOCOL_1_7_10) closeSession.run();
+        else statusRequest = handshake.nextState() == MinecraftHandshake.NextState.STATUS;
+    }
+
+    private void receiveStatus(ByteBuf packet) {
+        int id = ProtocolVarInt.read(packet.duplicate());
+        if (id == 0) {
+            frontend.writeAndFlush(owner.serverListStatus().encode(frontend.alloc(), owner.onlineCount()));
+        } else if (id == 1) {
+            ByteBuf input = packet.duplicate();
+            ProtocolVarInt.read(input);
+            if (input.readableBytes() != 8) throw new IllegalArgumentException("bad status ping");
+            ByteBuf pong = frontend.alloc().buffer(9);
+            ProtocolVarInt.write(pong, 1);
+            pong.writeLong(input.readLong());
+            frontend.writeAndFlush(pong).addListener(ignored -> frontend.close());
+        } else closeSession.run();
+    }
+
+    private void receiveEncryptionResponse(ByteBuf packet) {
+        MinecraftEncryptionResponse response = MinecraftEncryptionResponse.decode(packet);
+        AuthenticatedEncryption accepted;
+        try {
+            accepted = OnlineModeCrypto.decrypt(owner.encryptionKeys().getPrivate(), encryptionRequest, response);
+        } catch (GeneralSecurityException failure) {
+            closeSession.run();
+            return;
+        }
+        byte[] secret = accepted.sharedSecret();
+        String serverHash = OnlineModeCrypto.serverHash(encryptionRequest.serverId(), secret,
+                encryptionRequest.publicKey());
+        encryptionRequest = null;
+        ChannelPipeline pipeline = frontend.pipeline();
+        try {
+            pipeline.addAfter("minecraft-frame-decoder", "minecraft-cipher-decoder", new MinecraftCipherDecoder(secret));
+            pipeline.addAfter("minecraft-cipher-decoder", "encrypted-frame-decoder",
+                    new MinecraftFrameDecoder(ProtocolProfile.minecraft1710(), true,
+                            ProtocolProfile.MAX_LOGIN_FRAME_BYTES));
+            pipeline.addBefore("minecraft-frame-encoder", "minecraft-cipher-encoder", new MinecraftCipherEncoder(secret));
+            pipeline.remove("minecraft-frame-decoder");
+        } finally {
+            Arrays.fill(secret, (byte) 0);
+        }
+        guardPendingLogin();
+        String clientIp = ((InetSocketAddress) frontend.remoteAddress()).getAddress().getHostAddress();
+        verification = owner.verifier().verify(loginStart.username(), serverHash, clientIp).toCompletableFuture();
+        verification.whenComplete((profile, failure) -> frontend.eventLoop().execute(() -> {
+            verification = null;
+            if (shuttingDown.getAsBoolean()) return;
+            if (failure != null || profile == null || profile.isEmpty()) {
+                closeSession.run();
+                return;
+            }
+            verifiedProfile = profile.get();
+            if (!verifiedProfile.username().equalsIgnoreCase(loginStart.username())) {
+                closeSession.run();
+                return;
+            }
+            authenticated.accept(verifiedProfile.uuid(), verifiedProfile.username());
+        }));
+    }
+}

proxy-core/src/main/java/dev/moonbridge/core/session/InitialRouter.java

@@ -0,0 +1,159 @@
+package dev.moonbridge.core.session;
+
+import dev.moonbridge.api.PlacementDecision;
+import dev.moonbridge.core.backend.BackendId;
+import dev.moonbridge.core.backend.BackendView;
+import dev.moonbridge.core.session.channel.SessionChannels;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import net.kyori.adventure.text.Component;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+
+/**
+ * Picks the player's first backend: turns a placement decision into an ordered candidate list and dials
+ * it until one connects. Retries only before a backend has received any Minecraft handshake/login bytes.
+ * Confined to the frontend event loop.
+ */
+final class InitialRouter {
+    private static final Logger LOGGER = LoggerFactory.getLogger(InitialRouter.class);
+
+    /** Outcome callbacks; {@code connected} gets a live channel that nothing has been written to yet. */
+    interface Listener {
+        void connected(Channel backend, BackendView target);
+        void reject(Component reason);
+        void close();
+    }
+
+    private final ProxySessionListener owner;
+    private final Channel frontend;
+    private final BooleanSupplier shuttingDown;
+    private final Listener listener;
+    private List<String> candidates = List.of();
+    private int nextCandidate;
+    private long deadlineNanos;
+    private String username;
+    private Channel dialing;
+
+    InitialRouter(ProxySessionListener owner, Channel frontend, BooleanSupplier shuttingDown, Listener listener) {
+        this.owner = owner;
+        this.frontend = frontend;
+        this.shuttingDown = shuttingDown;
+        this.listener = listener;
+    }
+
+    /** Call when placement starts; the whole routing budget runs from here. */
+    void startClock() { deadlineNanos = System.nanoTime() + owner.placementTimeout().toNanos(); }
+
+    void route(String username, Optional<PlacementDecision> decision) {
+        this.username = username;
+        if (decision.isPresent()) {
+            if (decision.get() instanceof PlacementDecision.Reject rejected) {
+                listener.reject(rejected.reason());
+                return;
+            }
+            if (!(decision.get() instanceof PlacementDecision.Select selected)) {
+                listener.close();
+                return;
+            }
+            candidates = selected.backendNames();
+        } else {
+            candidates = owner.initialServers();
+        }
+        nextCandidate = 0;
+        if (candidates.isEmpty()) {
+            reject("No entry servers are configured.");
+            return;
+        }
+        dialNext();
+    }
+
+    /** Closes an in-flight dial that never became the session's backend. */
+    void cancel() {
+        Channel channel = dialing;
+        if (channel != null) channel.close();
+    }
+
+    private void reject(String reason) { listener.reject(Component.text(reason)); }
+
+    private void dialNext() {
+        if (shuttingDown.getAsBoolean()) return;
+        while (nextCandidate < candidates.size()) {
+            long remaining = deadlineNanos - System.nanoTime();
+            if (remaining <= 0) {
+                reject("Initial server routing timed out.");
+                return;
+            }
+            BackendView target = owner.catalog().find(new BackendId(candidates.get(nextCandidate++))).orElse(null);
+            if (target == null || !SessionChannels.isTcpAddress(target.address())) continue;
+            dial(target, remaining);
+            return;
+        }
+        reject("No entry server could be reached.");
+    }
+
+    private void dial(BackendView target, long remainingNanos) {
+        // The session is not attached to this channel until the attempt has won. A failed
+        // dial's inactive/exception callbacks must not close a later connection.
+        Bootstrap bootstrap = SessionChannels.backendBootstrap(frontend, owner.transport(), owner.backendResolver(),
+                (int) Math.max(1, Math.min(5000, TimeUnit.NANOSECONDS.toMillis(remainingNanos))), false,
+                pipeline -> { });
+        ChannelFuture connect;
+        try {
+            connect = bootstrap.connect(SessionChannels.socketAddress(target.address()));
+        } catch (RuntimeException failure) {
+            LOGGER.debug("Initial backend dial could not start for {} to {}", username, target.address(), failure);
+            dialNext();
+            return;
+        }
+        dialing = connect.channel();
+        connect.addListener(ignored -> finish(connect, target));
+    }
+
+    private void finish(ChannelFuture connect, BackendView target) {
+        if (!frontend.eventLoop().inEventLoop()) {
+            try {
+                frontend.eventLoop().execute(() -> finish(connect, target));
+            } catch (RejectedExecutionException shutdown) {
+                connect.channel().close();
+            }
+            return;
+        }
+        if (shuttingDown.getAsBoolean() || dialing != connect.channel()) {
+            connect.channel().close();
+            return;
+        }
+        if (!connect.isSuccess()) {
+            LOGGER.debug("Initial backend connection failed for player {} to {}", username, target.address(),
+                    connect.cause());
+            retry(connect);
+            return;
+        }
+        BackendView current = owner.catalog().find(target.handle().id()).orElse(null);
+        if (current == null || !current.handle().equals(target.handle())
+                || !current.address().equals(target.address())) {
+            retry(connect);
+            return;
+        }
+        if (deadlineNanos - System.nanoTime() <= 0) {
+            reject("Initial server routing timed out.");
+            return;
+        }
+        dialing = null;
+        candidates = List.of();
+        listener.connected(connect.channel(), current);
+    }
+
+    private void retry(ChannelFuture connect) {
+        dialing = null;
+        connect.channel().close();
+        dialNext();
+    }
+}

proxy-core/src/main/java/dev/moonbridge/core/session/ProxySessionListener.java

@@ -1,5 +1,7 @@
 package dev.moonbridge.core.session;
 
+import dev.moonbridge.core.session.channel.SessionFrameEncoder;
+
 import dev.moonbridge.api.PlacementDecision;
 import dev.moonbridge.api.event.Event;
 import dev.moonbridge.api.event.ConnectionAdmissionEvent;

proxy-core/src/main/java/dev/moonbridge/core/session/channel/SessionChannels.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.channel;
 
 import dev.moonbridge.core.net.NetworkTransport;
 import dev.moonbridge.core.protocol.MinecraftFrameDecoder;
@@ -17,19 +17,19 @@ import java.net.URI;
 import java.util.function.Consumer;
 
 /** Backend dialing and pipeline lookups shared by initial login and transfer. */
-final class SessionChannels {
-    static final String FRAME_DECODER = "minecraft-frame-decoder";
-    static final String ENCRYPTED_FRAME_DECODER = "encrypted-frame-decoder";
-    static final String FRAME_ENCODER = "minecraft-frame-encoder";
+public final class SessionChannels {
+    public static final String FRAME_DECODER = "minecraft-frame-decoder";
+    public static final String ENCRYPTED_FRAME_DECODER = "encrypted-frame-decoder";
+    public static final String FRAME_ENCODER = "minecraft-frame-encoder";
 
     private SessionChannels() { }
 
-    static boolean isTcpAddress(URI address) {
+    public static boolean isTcpAddress(URI address) {
         return "tcp".equalsIgnoreCase(address.getScheme()) && address.getHost() != null
                 && address.getPort() >= 1 && address.getPort() <= 65535;
     }
 
-    static InetSocketAddress socketAddress(URI address) {
+    public static InetSocketAddress socketAddress(URI address) {
         String host = address.getHost();
         if (host.startsWith("[") && host.endsWith("]")) {
             host = host.substring(1, host.length() - 1);
@@ -42,7 +42,7 @@ final class SessionChannels {
     }
 
     /** Dials on the player's event loop so session state stays confined to one thread. */
-    static Bootstrap backendBootstrap(Channel frontend, NetworkTransport transport,
+    public static Bootstrap backendBootstrap(Channel frontend, NetworkTransport transport,
                                       AddressResolverGroup<InetSocketAddress> resolver,
                                       int connectTimeoutMillis, boolean autoRead,
                                       Consumer<ChannelPipeline> extraHandlers) {
@@ -62,19 +62,19 @@ final class SessionChannels {
                 });
     }
 
-    static void installCodecs(ChannelPipeline pipeline) {
+    public static void installCodecs(ChannelPipeline pipeline) {
         pipeline.addLast(FRAME_DECODER, new MinecraftFrameDecoder(ProtocolProfile.minecraft1710(), true));
         pipeline.addLast(FRAME_ENCODER, new SessionFrameEncoder());
     }
 
     /** Name of the active frame decoder: the encrypted one replaces the plain one in online mode. */
-    static String frameDecoderName(ChannelPipeline pipeline) {
+    public static String frameDecoderName(ChannelPipeline pipeline) {
         if (pipeline.get(ENCRYPTED_FRAME_DECODER) != null) return ENCRYPTED_FRAME_DECODER;
         if (pipeline.get(FRAME_DECODER) != null) return FRAME_DECODER;
         return null;
     }
 
-    static MinecraftFrameDecoder frameDecoder(ChannelPipeline pipeline, String failure) {
+    public static MinecraftFrameDecoder frameDecoder(ChannelPipeline pipeline, String failure) {
         String name = frameDecoderName(pipeline);
         if (name == null || !(pipeline.get(name) instanceof MinecraftFrameDecoder decoder)) {
             throw new IllegalStateException(failure);

proxy-core/src/main/java/dev/moonbridge/core/session/channel/SessionFrameEncoder.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.channel;
 
 import dev.moonbridge.core.protocol.MinecraftFrameEncoder;
 import dev.moonbridge.core.protocol.ProtocolProfile;
@@ -8,7 +8,7 @@ import io.netty.handler.codec.MessageToMessageEncoder;
 
 import java.util.List;
 
-final class SessionFrameEncoder extends MessageToMessageEncoder<ByteBuf> {
+public final class SessionFrameEncoder extends MessageToMessageEncoder<ByteBuf> {
     @Override protected void encode(ChannelHandlerContext ctx, ByteBuf message, List<Object> output) {
         output.add(MinecraftFrameEncoder.encode(ctx.alloc(), message, ProtocolProfile.minecraft1710()));
     }

proxy-core/src/main/java/dev/moonbridge/core/session/play/CommandRateLimiter.java

@@ -1,25 +1,25 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import java.util.concurrent.TimeUnit;
 
 /** Token bucket for proxy-owned player commands; confined to the session event loop. */
-final class CommandRateLimiter {
-    enum Decision { ADMIT, DENY, DENY_WITH_NOTICE }
+public final class CommandRateLimiter {
+    public enum Decision { ADMIT, DENY, DENY_WITH_NOTICE }
 
-    static final int BURST = 10;
-    static final long TOKEN_NANOS = TimeUnit.MILLISECONDS.toNanos(200);
-    static final long NOTICE_NANOS = TimeUnit.SECONDS.toNanos(2);
+    public static final int BURST = 10;
+    public static final long TOKEN_NANOS = TimeUnit.MILLISECONDS.toNanos(200);
+    public static final long NOTICE_NANOS = TimeUnit.SECONDS.toNanos(2);
 
     private int tokens = BURST;
     private long refillNanos;
     private long lastNoticeNanos;
     private boolean noticeSent;
 
-    CommandRateLimiter(long nowNanos) {
+    public CommandRateLimiter(long nowNanos) {
         refillNanos = nowNanos;
     }
 
-    Decision admit(long nowNanos) {
+    public Decision admit(long nowNanos) {
         long elapsed = nowNanos - refillNanos;
         if (elapsed >= TOKEN_NANOS) {
             long replenished = Math.min(BURST, elapsed / TOKEN_NANOS);

proxy-core/src/main/java/dev/moonbridge/core/session/play/FrameTransformHandler.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import dev.moonbridge.core.relay.RawRelay;
 import io.netty.buffer.ByteBuf;
@@ -11,13 +11,13 @@ import io.netty.util.ReferenceCountUtil;
  * An inbound PLAY handler that sees one complete frame at a time and owns its reference counting, so
  * subclasses never call {@code release()}: {@link #transform} borrows the frame and answers what to forward.
  */
-abstract class FrameTransformHandler extends ChannelInboundHandlerAdapter {
+public abstract class FrameTransformHandler extends ChannelInboundHandlerAdapter {
     /** Returned by {@link #transform} when the frame was consumed and the relay resumes on its own later. */
-    static final ByteBuf CONSUMED = Unpooled.EMPTY_BUFFER;
+    public static final ByteBuf CONSUMED = Unpooled.EMPTY_BUFFER;
 
     private final Runnable closeSession;
 
-    FrameTransformHandler(Runnable closeSession) {
+    public FrameTransformHandler(Runnable closeSession) {
         this.closeSession = closeSession;
     }
 

proxy-core/src/main/java/dev/moonbridge/core/session/play/KeepAliveBridge.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
@@ -9,18 +9,18 @@ import java.util.Map;
 import java.util.concurrent.ThreadLocalRandom;
 
 /** Keeps connection-local PLAY keep-alive replies on the backend that issued them. */
-final class KeepAliveBridge extends FrameTransformHandler {
-    static final class State {
+public final class KeepAliveBridge extends FrameTransformHandler {
+    public static final class State {
         private static final int MAX_PENDING = 1024;
         private int nextClientId = ThreadLocalRandom.current().nextInt();
         private final Map<Integer, Integer> pending = new HashMap<>();
 
-        void switchBackend() {
+        public void switchBackend() {
             pending.clear();
         }
 
         /** Transforms a packet body while the login frame encoder is still installed. */
-        ByteBuf body(ByteBufAllocator allocator, ByteBuf packet, boolean fromFrontend) {
+        public ByteBuf body(ByteBufAllocator allocator, ByteBuf packet, boolean fromFrontend) {
             int start = packet.readerIndex();
             if (packet.readableBytes() == 0 || packet.getUnsignedByte(start) != 0) return packet;
             if (packet.readableBytes() != 5) throw new IllegalArgumentException("invalid keep-alive packet");
@@ -30,7 +30,7 @@ final class KeepAliveBridge extends FrameTransformHandler {
         }
 
         /** Transforms a complete length-prefixed PLAY frame without copying ordinary packets. */
-        ByteBuf frame(ByteBufAllocator allocator, ByteBuf packet, boolean fromFrontend) {
+        public ByteBuf frame(ByteBufAllocator allocator, ByteBuf packet, boolean fromFrontend) {
             int start = packet.readerIndex();
             if (packet.readableBytes() < 2 || packet.getUnsignedByte(start) != 5
                     || packet.getUnsignedByte(start + 1) != 0) return packet;
@@ -60,7 +60,7 @@ final class KeepAliveBridge extends FrameTransformHandler {
     private final boolean fromFrontend;
     private final Runnable closeSession;
 
-    KeepAliveBridge(State state, boolean fromFrontend, Runnable closeSession) {
+    public KeepAliveBridge(State state, boolean fromFrontend, Runnable closeSession) {
         super(closeSession);
         this.state = state;
         this.fromFrontend = fromFrontend;

proxy-core/src/main/java/dev/moonbridge/core/session/play/PlayObservation.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import dev.moonbridge.core.protocol.Minecraft1710PlayPackets;
 import dev.moonbridge.core.protocol.ProtocolProfile;
@@ -9,7 +9,7 @@ import java.util.OptionalInt;
 import java.util.concurrent.CompletableFuture;
 
 /** PLAY facts for a session; callers run on the player's event loop. */
-final class PlayObservation implements AutoCloseable {
+public final class PlayObservation implements AutoCloseable {
     private static final int MAX_FRAME_BYTES = ProtocolProfile.minecraft1710().maxFrameBytes();
     private final CompletableFuture<Void> ready = new CompletableFuture<>();
     private boolean joinSeen;
@@ -20,7 +20,7 @@ final class PlayObservation implements AutoCloseable {
     private OptionalInt forgeDimensionOverride = OptionalInt.empty();
     private OptionalInt entityId = OptionalInt.empty();
     /** Called after MinecraftFrameDecoder with one complete, length-prefixed frame. */
-    void observeFrame(boolean clientbound, ByteBuf frame) {
+    public void observeFrame(boolean clientbound, ByteBuf frame) {
         if (ready.isDone()) return;
         ByteBuf packet = frame.duplicate();
         int length = ProtocolVarInt.read(packet);
@@ -30,7 +30,7 @@ final class PlayObservation implements AutoCloseable {
         observePacket(clientbound, packet);
     }
 
-    void observePacket(boolean clientbound, ByteBuf packet) {
+    public void observePacket(boolean clientbound, ByteBuf packet) {
         if (ready.isDone()) return;
         int id = ProtocolVarInt.read(packet.duplicate());
         if (clientbound) {
@@ -65,10 +65,10 @@ final class PlayObservation implements AutoCloseable {
         if (joinSeen && forgeSeen && backendComplete && clientComplete) ready.complete(null);
     }
 
-    CompletableFuture<Void> ready() { return ready; }
-    boolean forgeSeen() { return forgeSeen; }
-    OptionalInt dimension() { return dimension; }
-    OptionalInt entityId() { return entityId; }
+    public CompletableFuture<Void> ready() { return ready; }
+    public boolean forgeSeen() { return forgeSeen; }
+    public OptionalInt dimension() { return dimension; }
+    public OptionalInt entityId() { return entityId; }
 
     @Override public void close() {
         if (!ready.isDone()) ready.completeExceptionally(new IllegalStateException("session closed"));

proxy-core/src/main/java/dev/moonbridge/core/session/play/PlayerCommandInterceptor.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import dev.moonbridge.core.protocol.Minecraft1710PlayPackets;
 import io.netty.buffer.ByteBuf;
@@ -7,10 +7,10 @@ import io.netty.channel.ChannelHandlerContext;
 import java.util.function.Predicate;
 
 /** Consumes only registered proxy commands; all other PLAY frames keep their wire bytes. */
-final class PlayerCommandInterceptor extends FrameTransformHandler {
+public final class PlayerCommandInterceptor extends FrameTransformHandler {
     private final Predicate<String> dispatch;
 
-    PlayerCommandInterceptor(Predicate<String> dispatch, Runnable closeSession) {
+    public PlayerCommandInterceptor(Predicate<String> dispatch, Runnable closeSession) {
         super(closeSession);
         this.dispatch = dispatch;
     }

proxy-core/src/main/java/dev/moonbridge/core/session/play/TabCompletionBridge.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import dev.moonbridge.core.protocol.MinecraftTabCompletion;
 import dev.moonbridge.core.relay.RawRelay;
@@ -20,7 +20,7 @@ import java.util.function.Function;
 import java.util.function.Predicate;
 
 /** One backend generation, confined to the shared frontend/backend event loop. */
-final class TabCompletionBridge implements AutoCloseable {
+public final class TabCompletionBridge implements AutoCloseable {
     private final Channel frontend;
     private final Function<String, List<String>> roots;
     private final Function<String, Optional<CompletionStage<List<String>>>> complete;
@@ -44,13 +44,13 @@ final class TabCompletionBridge implements AutoCloseable {
     private ScheduledFuture<?> timeout;
     private CompletableFuture<?> pluginResult;
 
-    TabCompletionBridge(Channel frontend, Function<String, List<String>> roots,
+    public TabCompletionBridge(Channel frontend, Function<String, List<String>> roots,
                         Function<String, Optional<CompletionStage<List<String>>>> complete,
                         Runnable closeSession) {
         this(frontend, roots, complete, closeSession, ignored -> true);
     }
 
-    TabCompletionBridge(Channel frontend, Function<String, List<String>> roots,
+    public TabCompletionBridge(Channel frontend, Function<String, List<String>> roots,
                         Function<String, Optional<CompletionStage<List<String>>>> complete,
                         Runnable closeSession, Predicate<String> visibleRoot) {
         this.frontend = frontend;
@@ -60,8 +60,8 @@ final class TabCompletionBridge implements AutoCloseable {
         this.visibleRoot = visibleRoot;
     }
 
-    ChannelInboundHandlerAdapter frontendHandler() { return new Handler(true); }
-    ChannelInboundHandlerAdapter backendHandler() { return new Handler(false); }
+    public ChannelInboundHandlerAdapter frontendHandler() { return new Handler(true); }
+    public ChannelInboundHandlerAdapter backendHandler() { return new Handler(false); }
 
     private final class Handler extends ChannelInboundHandlerAdapter {
         private final boolean fromClient;

proxy-core/src/main/java/dev/moonbridge/core/session/play/TransitionBuffer.java

@@ -1,25 +1,25 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import io.netty.buffer.ByteBuf;
 
 import java.util.ArrayDeque;
 
 /** Bounded queue of frames read while the relay starts; it owns every buffer it holds until polled. */
-final class TransitionBuffer {
-    record Frame(boolean fromFrontend, ByteBuf payload) { }
+public final class TransitionBuffer {
+    public record Frame(boolean fromFrontend, ByteBuf payload) { }
 
     private final int maxFrames;
     private final int maxBytes;
     private final ArrayDeque<Frame> frames = new ArrayDeque<>();
     private int bytes;
 
-    TransitionBuffer(int maxFrames, int maxBytes) {
+    public TransitionBuffer(int maxFrames, int maxBytes) {
         this.maxFrames = maxFrames;
         this.maxBytes = maxBytes;
     }
 
     /** Retains a duplicate of {@code packet}; returns false, retaining nothing, when a bound would be exceeded. */
-    boolean add(boolean fromFrontend, ByteBuf packet) {
+    public boolean add(boolean fromFrontend, ByteBuf packet) {
         int size = packet.readableBytes();
         if (frames.size() >= maxFrames || bytes + size > maxBytes) return false;
         frames.addLast(new Frame(fromFrontend, packet.retainedDuplicate()));
@@ -27,19 +27,19 @@ final class TransitionBuffer {
         return true;
     }
 
-    boolean isEmpty() { return frames.isEmpty(); }
+    public boolean isEmpty() { return frames.isEmpty(); }
 
-    int size() { return frames.size(); }
+    public int size() { return frames.size(); }
 
     /** Transfers ownership of the oldest frame to the caller, or returns null when empty. */
-    Frame poll() {
+    public Frame poll() {
         Frame frame = frames.pollFirst();
         if (frame != null) bytes -= frame.payload().readableBytes();
         return frame;
     }
 
     /** Releases everything still queued. */
-    void release() {
+    public void release() {
         Frame frame;
         while ((frame = frames.pollFirst()) != null) frame.payload().release();
         bytes = 0;

proxy-core/src/main/java/dev/moonbridge/core/session/play/TransitionFrames.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import dev.moonbridge.core.protocol.ByteBufs;
 import dev.moonbridge.core.protocol.Minecraft1710EntityIds;
@@ -8,14 +8,14 @@ import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.Channel;
 
 /** Maps frames buffered during relay start-up before they are written to their destination. */
-final class TransitionFrames {
+public final class TransitionFrames {
     private TransitionFrames() { }
 
     /**
      * Takes ownership of {@code payload} and returns the buffer to send, or null when the frame is
      * swallowed. The payload is released on every failure, so callers never touch it after an exception.
      */
-    static ByteBuf map(PlayObservation observation, KeepAliveBridge.State keepAlives, boolean fromFrontend,
+    public static ByteBuf map(PlayObservation observation, KeepAliveBridge.State keepAlives, boolean fromFrontend,
                        ByteBuf payload, Channel target) {
         try {
             if (observation != null) observation.observePacket(!fromFrontend, payload);
@@ -33,7 +33,7 @@ final class TransitionFrames {
      * Frames {@code mapped} (borrowed), swaps the player's entity ids when the replacement backend assigned a
      * different one, and appends the result to {@code output}. Every temporary buffer is released.
      */
-    static void appendClientbound(ByteBuf output, ByteBufAllocator allocator, ByteBuf mapped,
+    public static void appendClientbound(ByteBuf output, ByteBufAllocator allocator, ByteBuf mapped,
                                   Integer serverEntityId, Integer clientEntityId) {
         ByteBufs.use(Minecraft1710PlayPackets.frame(allocator, mapped), frame -> {
             ByteBuf outgoing = serverEntityId == null ? frame

proxy-core/src/main/java/dev/moonbridge/core/session/transfer/PendingTransfer.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.transfer;
 
 import dev.moonbridge.api.TransferResult;
 

proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferAttempt.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.transfer;
 
 import dev.moonbridge.api.TransferResult;
 import dev.moonbridge.api.event.TransferContext;

proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferCandidate.java

@@ -1,4 +1,6 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.transfer;
+
+import dev.moonbridge.core.session.play.PlayObservation;
 
 import dev.moonbridge.core.protocol.Minecraft1710PlayPackets;
 import dev.moonbridge.core.protocol.MinecraftLoginSuccess;
@@ -17,8 +19,8 @@ import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
 /** Logs into a replacement backend while the existing player session keeps its old backend. */
-final class TransferCandidate extends ChannelInboundHandlerAdapter {
-    interface Listener {
+public final class TransferCandidate extends ChannelInboundHandlerAdapter {
+    public interface Listener {
         void ready(TransferCandidate candidate);
         void failed(TransferCandidate candidate, String reason);
     }
@@ -38,11 +40,11 @@ final class TransferCandidate extends ChannelInboundHandlerAdapter {
     private State state = State.LOGIN;
     private int queuedBytes;
 
-    TransferCandidate(UUID expectedId, String username, Listener listener) {
+    public TransferCandidate(UUID expectedId, String username, Listener listener) {
         this(expectedId, username, listener, false);
     }
 
-    TransferCandidate(UUID expectedId, String username, Listener listener, boolean deferLogin) {
+    public TransferCandidate(UUID expectedId, String username, Listener listener, boolean deferLogin) {
         this.expectedId = expectedId;
         this.username = username;
         this.listener = listener;
@@ -55,7 +57,7 @@ final class TransferCandidate extends ChannelInboundHandlerAdapter {
     }
 
     /** Coordinated waits must not consume the destination's login budget. */
-    void loginStarted() {
+    public void loginStarted() {
         if (loginStarted) return;
         if (context == null || !context.executor().inEventLoop()) {
             throw new IllegalStateException("candidate login must start on its event loop");
@@ -141,23 +143,23 @@ final class TransferCandidate extends ChannelInboundHandlerAdapter {
         else fail(ctx, "replacement backend failed");
     }
 
-    Minecraft1710PlayPackets.JoinGame joinGame() { return joinGame; }
-    PlayObservation observation() { return observation; }
+    public Minecraft1710PlayPackets.JoinGame joinGame() { return joinGame; }
+    public PlayObservation observation() { return observation; }
 
-    List<ByteBuf> takeQueuedPackets() {
+    public List<ByteBuf> takeQueuedPackets() {
         var packets = new ArrayList<>(queued);
         queued.clear();
         queuedBytes = 0;
         return packets;
     }
 
-    void handOff() {
+    public void handOff() {
         if (state != State.READY) throw new IllegalStateException("replacement backend is not ready");
         state = State.HANDED_OFF;
         deadline.cancel(false);
     }
 
-    void close() {
+    public void close() {
         if (deadline != null) deadline.cancel(false);
         ByteBuf packet;
         while ((packet = queued.pollFirst()) != null) packet.release();

proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferCoordinator.java

@@ -0,0 +1,629 @@
+package dev.moonbridge.core.session.transfer;
+
+import dev.moonbridge.api.ServerView;
+import dev.moonbridge.api.TransferResult;
+import dev.moonbridge.api.TransferStatus;
+import dev.moonbridge.api.event.TransferContext;
+import dev.moonbridge.api.event.TransferPreparingEvent;
+import dev.moonbridge.core.backend.BackendId;
+import dev.moonbridge.core.backend.BackendView;
+import dev.moonbridge.core.event.PreparedTransfer;
+import dev.moonbridge.core.protocol.ByteBufs;
+import dev.moonbridge.core.protocol.LoginStart;
+import dev.moonbridge.core.protocol.ProtocolProfile;
+import dev.moonbridge.core.protocol.Minecraft1710PlayPackets;
+import dev.moonbridge.core.relay.RawRelay;
+import dev.moonbridge.core.session.channel.SessionChannels;
+import dev.moonbridge.core.session.play.KeepAliveBridge;
+import dev.moonbridge.core.session.play.PlayObservation;
+import dev.moonbridge.core.session.play.TransitionFrames;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelPipeline;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Moves a connected player to another backend while the old link keeps serving: dials and logs into the
+ * candidate, buffers both directions, swaps relays and rewrites the player's entity ids. Everything runs on
+ * the player's frontend event loop; the session is reached only through {@link TransferHost}.
+ */
+public final class TransferCoordinator {
+    private static final Logger LOGGER = LoggerFactory.getLogger(TransferCoordinator.class);
+    private static final Duration FORGE_TRANSFER_HANDSHAKE_TIMEOUT = Duration.ofSeconds(30);
+
+    private final TransferHost host;
+    private final Channel frontend;
+    private PendingTransfer pendingTransfer;
+    private TransferAttempt transfer;
+    private TransferFrameHandler.State frameState;
+    private Integer clientEntityId;
+    private boolean clientFmlAwaitingServerHello;
+
+    public TransferCoordinator(TransferHost host) {
+        this.host = host;
+        this.frontend = host.frontend();
+    }
+
+    /** True while a transfer is queued or running; other writers to the client must wait. */
+    public boolean busy() { return transfer != null || pendingTransfer != null; }
+
+    public CompletionStage<TransferResult> transferTo(String backendName) {
+        CompletableFuture<TransferResult> result = new CompletableFuture<>();
+        Runnable command = () -> {
+            try {
+                beginTransfer(backendName, result);
+            } catch (RuntimeException failure) {
+                result.completeExceptionally(failure);
+                host.closeSession();
+            }
+        };
+        if (frontend.eventLoop().inEventLoop()) command.run();
+        else {
+            try {
+                frontend.eventLoop().execute(command);
+            } catch (RejectedExecutionException shutdown) {
+                result.complete(TransferResult.of(TransferStatus.PLAYER_NOT_CONNECTED));
+            }
+        }
+        return result;
+    }
+
+    private void beginTransfer(String backendName, CompletableFuture<TransferResult> result) {
+        if (host.closed() || host.disconnecting() || !host.published()) {
+            result.complete(TransferResult.of(TransferStatus.PLAYER_NOT_CONNECTED));
+            return;
+        }
+        if (transfer != null || pendingTransfer != null) {
+            result.complete(TransferResult.failed("a backend transfer is already in progress"));
+            return;
+        }
+        if (host.selected().handle().id().value().equals(backendName)) {
+            BackendView current = host.catalog().find(host.selected().handle().id()).orElse(null);
+            if (current != null && current.address().equals(host.selected().address())) {
+                result.complete(TransferResult.of(TransferStatus.NETWORK_READY));
+                return;
+            }
+        }
+        if (host.observation().ready().isCompletedExceptionally()) {
+            result.complete(TransferResult.failed("backend login or Forge negotiation failed"));
+            return;
+        }
+        if (host.relay() == null || !host.observation().ready().isDone()) {
+            PendingTransfer waiting = new PendingTransfer(backendName, result);
+            pendingTransfer = waiting;
+            waiting.deadline = frontend.eventLoop().schedule(() -> {
+                if (pendingTransfer == waiting) {
+                    pendingTransfer = null;
+                    result.complete(TransferResult.failed("backend login or Forge negotiation timed out"));
+                }
+            }, 15, TimeUnit.SECONDS);
+            return;
+        }
+        BackendView target;
+        try {
+            target = host.catalog().find(new BackendId(backendName)).orElse(null);
+        } catch (IllegalArgumentException invalidName) {
+            target = null;
+        }
+        if (target == null || !SessionChannels.isTcpAddress(target.address())) {
+            result.complete(TransferResult.of(TransferStatus.SERVER_UNAVAILABLE));
+            return;
+        }
+        TransferAttempt attempt = new TransferAttempt(target, result, host.relay());
+        attempt.preparation = host.selectPreparation();
+        if (attempt.preparation != null) {
+            Duration totalBudget = host.eventTimeout().multipliedBy(2).plusSeconds(20)
+                    .plus(host.cutoverTimeout()).plus(FORGE_TRANSFER_HANDSHAKE_TIMEOUT);
+            attempt.context = new TransferContext(UUID.randomUUID(), host.identity(), serverView(host.selected()), serverView(target),
+                    host.selected().handle().generation(), target.handle().generation(),
+                    host.selected().owner().instanceGeneration(), target.owner().instanceGeneration(),
+                    Instant.now().plus(totalBudget));
+            attempt.totalDeadline = frontend.eventLoop().schedule(() -> {
+                if (currentAttempt(attempt)) failTransfer(attempt, "coordinated transfer timed out");
+            }, totalBudget.toNanos(), TimeUnit.NANOSECONDS);
+        }
+        transfer = attempt;
+        LOGGER.debug("Starting replacement backend connection for player {} to {}",
+                host.view().username(), target.address());
+        attempt.candidate = new TransferCandidate(host.identity().playerId(), host.view().username(), new TransferCandidate.Listener() {
+            @Override public void ready(TransferCandidate candidate) { candidateReady(attempt); }
+            @Override public void failed(TransferCandidate candidate, String reason) { failTransfer(attempt, reason); }
+        }, attempt.preparation != null);
+        try {
+            Bootstrap bootstrap = SessionChannels.backendBootstrap(frontend, host.transport(),
+                    host.resolver(), 5000, true,
+                    pipeline -> pipeline.addLast("transfer-candidate", attempt.candidate));
+            ChannelFuture connect = bootstrap.connect(SessionChannels.socketAddress(attempt.target.address()));
+            attempt.channel = connect.channel();
+            connect.addListener(future -> {
+                if (attempt.finished || host.closed() || host.disconnecting()) { connect.channel().close(); return; }
+                if (!future.isSuccess()) {
+                    LOGGER.debug("Replacement backend connection failed for player {} to {}",
+                            host.view().username(), attempt.target.address(), future.cause());
+                    failTransfer(attempt, "could not connect to replacement backend");
+                    return;
+                }
+                LOGGER.debug("Replacement backend TCP connection established for player {} to {}",
+                        host.view().username(), attempt.target.address());
+                if (attempt.preparation == null) writeBackendLogin(connect.channel());
+                else prepareTransfer(attempt);
+            });
+        } catch (RuntimeException failure) {
+            failTransfer(attempt, "could not start replacement backend connection");
+        }
+    }
+
+    private static ServerView serverView(BackendView backend) {
+        return new ServerView(backend.handle().id().value(), backend.address(), backend.tags(), backend.metadata());
+    }
+
+    private boolean currentAttempt(TransferAttempt attempt) {
+        return transfer == attempt && !attempt.finished && !host.closed() && !host.disconnecting();
+    }
+
+    private boolean targetStillCurrent(TransferAttempt attempt) {
+        BackendView current = host.catalog().find(attempt.target.handle().id()).orElse(null);
+        return current != null && current.handle().equals(attempt.target.handle())
+                && current.owner().equals(attempt.target.owner()) && current.address().equals(attempt.target.address())
+                && attempt.channel != null && attempt.channel.isActive();
+    }
+
+    private void prepareTransfer(TransferAttempt attempt) {
+        try {
+            CompletionStage<PreparedTransfer> preparation = attempt.preparation.prepare(
+                    new TransferPreparingEvent(attempt.context));
+            attempt.coordination = preparation.toCompletableFuture();
+            preparation.whenComplete((prepared, failure) -> frontend.eventLoop().execute(() -> {
+                if (!currentAttempt(attempt)) return;
+                attempt.coordination = null;
+                if (failure != null || prepared == null) {
+                    LOGGER.debug("Transfer preparation failed for {}", host.view().username(), failure);
+                    failTransfer(attempt, "transfer preparation failed");
+                } else if (!targetStillCurrent(attempt)) {
+                    failTransfer(attempt, "replacement backend registration changed");
+                } else if (prepared.requiresSourceRelease()) {
+                    releaseTransferSource(attempt, prepared);
+                } else {
+                    writeBackendLogin(attempt.channel);
+                }
+            }));
+        } catch (RuntimeException failure) {
+            failTransfer(attempt, "could not prepare backend transfer");
+        }
+    }
+
+    private void releaseTransferSource(TransferAttempt attempt, PreparedTransfer prepared) {
+        Channel source = host.backend();
+        pauseSource(attempt, source, "could not stop source gameplay for transfer",
+                "could not drain source relay", () -> {
+            if (!currentAttempt(attempt)) {
+                if (!host.closed() && !host.disconnecting()) resumeAfterFailedTransfer(attempt);
+                return;
+            }
+            if (clientHasPartialFrame()) {
+                failTransfer(attempt, "client packet was incomplete at the transfer boundary");
+                return;
+            }
+            if (!targetStillCurrent(attempt)) {
+                failTransfer(attempt, "replacement backend became unavailable");
+                return;
+            }
+            attempt.oldRelay.detach().whenComplete((removed, detachFailure) -> frontend.eventLoop().execute(() -> {
+                if (detachFailure != null) {
+                    boolean attached = frontend.pipeline().get("raw-relay") != null
+                            && source.pipeline().get("raw-relay") != null;
+                    failTransfer(attempt, "could not detach source relay");
+                    if (!attached) host.closeSession();
+                    return;
+                }
+                attempt.detached = true;
+                if (!currentAttempt(attempt) || !targetStillCurrent(attempt)) {
+                    failTransfer(attempt, "replacement backend closed during source release");
+                    host.closeSession();
+                    return;
+                }
+                // Disarm exact old-channel EOF before closing it; no destination exists yet.
+                attempt.sourceReleased = true;
+                attempt.clientBuffer.discardFrames();
+                attempt.oldBackendBuffer.discardFrames();
+                host.sourceReleased();
+                if (source.pipeline().get("transfer-frame-handler") != null) {
+                    source.pipeline().remove("transfer-frame-handler");
+                }
+                source.close().addListener(closedSource -> {
+                    if (!currentAttempt(attempt)) return;
+                    if (!closedSource.isSuccess()) {
+                        failTransfer(attempt, "could not close source connection");
+                        return;
+                    }
+                    confirmSourceReleased(attempt, prepared);
+                });
+            }));
+        });
+    }
+
+    private void confirmSourceReleased(TransferAttempt attempt, PreparedTransfer prepared) {
+        try {
+            CompletionStage<Void> confirmation = prepared.sourceClosed();
+            attempt.coordination = confirmation.toCompletableFuture();
+            confirmation.whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> {
+                if (!currentAttempt(attempt)) return;
+                attempt.coordination = null;
+                if (failure != null) {
+                    LOGGER.debug("Source release confirmation failed for {}", host.view().username(), failure);
+                    failTransfer(attempt, "source release confirmation failed");
+                } else if (!targetStillCurrent(attempt)) {
+                    failTransfer(attempt, "replacement backend registration changed");
+                } else writeBackendLogin(attempt.channel);
+            }));
+        } catch (RuntimeException failure) {
+            failTransfer(attempt, "could not confirm source release");
+        }
+    }
+
+    /** Starts the queued transfer once the relay is attached and the PLAY/Forge handshake has finished. */
+    public void startPending() {
+        PendingTransfer waiting = pendingTransfer;
+        if (host.disconnecting() || waiting == null || host.relay() == null || !host.observation().ready().isDone()) return;
+        pendingTransfer = null;
+        waiting.deadline.cancel(false);
+        beginTransfer(waiting.backendName, waiting.result);
+    }
+
+    private void writeBackendLogin(Channel channel) {
+        transfer.candidate.loginStarted();
+        ByteBuf handshakeBody = host.backendHandshake(
+                transfer.target.handle().id().value(), transfer.target.owner().instanceGeneration());
+        ByteBuf loginBody = new LoginStart(host.view().username()).encode(frontend.alloc(), ProtocolProfile.minecraft1710());
+        channel.write(handshakeBody);
+        channel.writeAndFlush(loginBody).addListener(write -> {
+            if (write.isSuccess()) {
+                LOGGER.debug("Replacement backend login frames flushed for player {} to {}",
+                        host.view().username(), channel.remoteAddress());
+            } else {
+                LOGGER.debug("Replacement backend login write failed for player {}",
+                        host.view().username(), write.cause());
+            }
+            if (!write.isSuccess() && transfer != null && transfer.channel == channel) {
+                failTransfer(transfer, "could not write replacement backend login");
+            }
+        });
+    }
+
+    private void candidateReady(TransferAttempt attempt) {
+        if (transfer != attempt || attempt.finished || host.closed() || host.disconnecting()) return;
+        if (attempt.preparation != null && !targetStillCurrent(attempt)) {
+            failTransfer(attempt, "replacement backend registration changed during login");
+            return;
+        }
+        attempt.cutoverDeadline = frontend.eventLoop().schedule(() -> {
+            if (transfer != attempt || attempt.result.isDone()) return;
+            attempt.result.complete(TransferResult.failed("replacement backend cutover timed out"));
+            host.closeSession();
+        }, host.cutoverTimeout().toNanos(), TimeUnit.NANOSECONDS);
+        if (attempt.sourceReleased) {
+            if (clientHasPartialFrame()) {
+                failTransfer(attempt, "client packet was incomplete at replacement cutover");
+                return;
+            }
+            activateCandidate(attempt);
+            return;
+        }
+        pauseSource(attempt, host.backend(), "could not buffer frames for transfer",
+                "old backend stopped during transfer", () -> {
+            if (attempt.finished || host.closed() || host.disconnecting()) {
+                if (host.disconnecting()) return;
+                resumeAfterFailedTransfer(attempt);
+                return;
+            }
+            if (clientHasPartialFrame()) {
+                failTransfer(attempt, "client packet was incomplete at the transfer boundary");
+                return;
+            }
+            if (!attempt.channel.isActive()) {
+                failTransfer(attempt, "replacement backend became unavailable");
+                return;
+            }
+            attempt.oldRelay.detach().whenComplete((removed, detachFailure) ->
+                    frontend.eventLoop().execute(() -> {
+                        if (detachFailure != null) {
+                            boolean oldRelayStillAttached = frontend.pipeline().get("raw-relay") != null
+                                    && host.backend().pipeline().get("raw-relay") != null;
+                            failTransfer(attempt, "could not detach old backend relay");
+                            if (!oldRelayStillAttached) host.closeSession();
+                            return;
+                        }
+                        if (host.closed() || host.disconnecting() || !attempt.channel.isActive()) {
+                            if (host.disconnecting()) return;
+                            failTransfer(attempt, "replacement backend closed during transfer");
+                            host.closeSession();
+                            return;
+                        }
+                        attempt.detached = true;
+                        activateCandidate(attempt);
+                    }));
+        });
+    }
+
+    /**
+     * Buffers both old-link directions and pauses the old host.relay(). {@code onPaused} runs on the session
+     * event loop after accepted writes drain; buffer or drain failures fail the attempt and close.
+     */
+    private void pauseSource(TransferAttempt attempt, Channel source, String bufferFailure, String drainFailure,
+                             Runnable onPaused) {
+        try {
+            attempt.clientBuffer = installTransferBuffer(frontend, "transfer-client-buffer");
+            attempt.oldBackendBuffer = installTransferBuffer(source, "transfer-old-backend-buffer");
+        } catch (RuntimeException failure) {
+            failTransfer(attempt, bufferFailure);
+            host.closeSession();
+            return;
+        }
+        attempt.pauseInProgress = true;
+        CompletionStage<Void> pause = attempt.oldRelay.pause();
+        frontend.eventLoop().execute(attempt.clientBuffer::readUntilRemoved);
+        source.eventLoop().execute(attempt.oldBackendBuffer::readUntilRemoved);
+        pause.whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> {
+            attempt.pauseInProgress = false;
+            if (failure != null) {
+                failTransfer(attempt, drainFailure);
+                host.closeSession();
+                return;
+            }
+            attempt.paused = true;
+            onPaused.run();
+        }));
+    }
+
+    private TransferFrameBuffer installTransferBuffer(Channel channel, String name) {
+        var buffer = new TransferFrameBuffer(host::closeSession);
+        channel.pipeline().addAfter(SessionChannels.frameDecoderName(channel.pipeline()), name, buffer);
+        return buffer;
+    }
+
+    private boolean clientHasPartialFrame() {
+        return SessionChannels.frameDecoder(frontend.pipeline(), "client frame decoder missing during transfer")
+                .hasPartialFrame();
+    }
+
+    private void activateCandidate(TransferAttempt attempt) {
+        // Keep READY ownership until removal. A decoder can emit more target frames after
+        // JoinGame in the same read batch; marking HANDED_OFF early would silently drop them.
+        host.swapHandshakeCodecs(attempt.channel, attempt.candidate::handOff).whenComplete((ignored, failure) ->
+                frontend.eventLoop().execute(() -> {
+                    if (failure != null || host.closed() || host.disconnecting() || !attempt.channel.isActive()) {
+                        if (host.disconnecting()) return;
+                        failTransfer(attempt, "replacement backend pipeline failed");
+                        host.closeSession();
+                        return;
+                    }
+                    installCandidateRelay(attempt);
+                }));
+    }
+
+    private void installCandidateRelay(TransferAttempt attempt) {
+        PlayObservation nextObservation = attempt.candidate.observation();
+        RawRelay.Link next;
+        try {
+            if (clientEntityId == null) clientEntityId = host.observation().entityId().orElseThrow();
+            attempt.frameState = new TransferFrameHandler.State(frontend, attempt.channel, nextObservation,
+                    clientEntityId, attempt.candidate.joinGame(), host::closeSession);
+            attempt.channel.pipeline().addLast("keep-alive-bridge",
+                    new KeepAliveBridge(host.keepAlives(), false, host::closeSession));
+            installTransferFrameHandlers(attempt.frameState, attempt.channel);
+            host.installTabCompletion(attempt.channel);
+            next = RawRelay.attach(frontend, attempt.channel,
+                    bytes -> nextObservation.observeFrame(false, bytes),
+                    bytes -> nextObservation.observeFrame(true, bytes));
+            RawRelay.Link observedLink = next;
+            nextObservation.ready().whenComplete((ignored, failure) -> observedLink.stopObserving());
+        } catch (RuntimeException failure) {
+            failTransfer(attempt, "could not attach replacement relay");
+            host.closeSession();
+            return;
+        }
+        next.ready().whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> {
+            if (failure != null || host.closed() || host.disconnecting()
+                    || (attempt.preparation != null && !targetStillCurrent(attempt))) {
+                if (host.disconnecting()) return;
+                failTransfer(attempt, "replacement relay was not ready");
+                host.closeSession();
+                return;
+            }
+            ByteBuf opening;
+            try {
+                host.keepAlives().switchBackend();
+                opening = transferOpening(attempt);
+            } catch (RuntimeException malformed) {
+                failTransfer(attempt, "could not prepare replacement world transition");
+                host.closeSession();
+                return;
+            }
+            frontend.writeAndFlush(opening).addListener(write -> frontend.eventLoop().execute(() -> {
+                if (!write.isSuccess() || host.closed() || host.disconnecting() || !attempt.channel.isActive()
+                        || (attempt.preparation != null && !targetStillCurrent(attempt))) {
+                    if (host.disconnecting()) return;
+                    failTransfer(attempt, "could not send replacement world transition");
+                    host.closeSession();
+                    return;
+                }
+                Channel oldBackend = host.backend();
+                Optional<String> previousServer = attempt.sourceReleased
+                        ? Optional.of(attempt.context.source().name()) : host.view().currentServer();
+                PlayObservation oldObservation = host.observation();
+                TransferFrameHandler.State oldFrameState = frameState;
+                frameState = attempt.frameState;
+                host.cutover(attempt.channel, attempt.target, nextObservation, next, previousServer);
+                if (oldBackend != null) oldBackend.close();
+                oldObservation.close();
+                if (oldFrameState != null) oldFrameState.close();
+                try {
+                    attempt.clientBuffer.drainAndRemove();
+                } catch (RuntimeException replayFailure) {
+                    host.closeSession();
+                    return;
+                }
+                if (host.closed()) return;
+                next.start();
+                if (host.closed()) return;
+                if (nextObservation.forgeSeen()) {
+                    attempt.cutoverDeadline.cancel(false);
+                    attempt.cutoverDeadline = frontend.eventLoop().schedule(() -> {
+                        if (transfer != attempt || attempt.finished || host.closed() || host.disconnecting()) return;
+                        failTransfer(attempt, "replacement Forge handshake timed out");
+                        host.closeSession();
+                    }, FORGE_TRANSFER_HANDSHAKE_TIMEOUT.toNanos(), TimeUnit.NANOSECONDS);
+                    CompletableFuture.allOf(nextObservation.ready(), attempt.frameState.worldReady())
+                            .whenComplete((negotiated, negotiationFailure) -> frontend.eventLoop().execute(() -> {
+                                if (transfer != attempt || attempt.finished || host.closed() || host.disconnecting()) return;
+                                if (negotiationFailure != null || !attempt.channel.isActive()) {
+                                    failTransfer(attempt, "replacement Forge handshake failed");
+                                    host.closeSession();
+                                } else finishTransfer(attempt);
+                            }));
+                } else finishTransfer(attempt);
+            }));
+        }));
+    }
+
+    private void finishTransfer(TransferAttempt attempt) {
+        if (transfer != attempt || attempt.finished || host.closed() || host.disconnecting()) return;
+        transfer = null;
+        attempt.finished = true;
+        attempt.cutoverDeadline.cancel(false);
+        cancelCoordination(attempt);
+        attempt.result.complete(TransferResult.of(TransferStatus.NETWORK_READY));
+    }
+
+    private void installTransferFrameHandlers(TransferFrameHandler.State state, Channel target) {
+        ChannelPipeline clientPipeline = frontend.pipeline();
+        if (clientPipeline.get("transfer-frame-handler") != null) clientPipeline.remove("transfer-frame-handler");
+        if (SessionChannels.frameDecoderName(clientPipeline) == null) {
+            throw new IllegalStateException("client frame decoder missing during transfer");
+        }
+        clientPipeline.addLast("transfer-frame-handler", new TransferFrameHandler(state, false));
+        target.pipeline().addLast("transfer-frame-handler", new TransferFrameHandler(state, true));
+    }
+
+    private ByteBuf transferOpening(TransferAttempt attempt) {
+        List<ByteBuf> queued = attempt.candidate.takeQueuedPackets();
+        try {
+            return ByteBufs.fill(frontend.alloc().buffer(), output -> {
+                boolean nextForge = attempt.candidate.observation().forgeSeen();
+                // FML's reset also restores the client's frozen registry. A Forge -> vanilla
+                // switch needs it even though the replacement server has no ServerHello.
+                // After that reset the client remains in HELLO until a later Forge switch.
+                if (!clientFmlAwaitingServerHello && (host.observation().forgeSeen() || nextForge)) {
+                    ByteBuf reset = Minecraft1710PlayPackets.forgeReset(frontend.alloc());
+                    try { output.writeBytes(reset); } finally { reset.release(); }
+                    clientFmlAwaitingServerHello = true;
+                }
+                if (nextForge) clientFmlAwaitingServerHello = false;
+                if (attempt.candidate.joinGame() != null) {
+                    int targetDimension = attempt.candidate.observation().dimension()
+                            .orElse(attempt.candidate.joinGame().dimension());
+                    ByteBuf respawns = Minecraft1710PlayPackets.respawnSequence(frontend.alloc(),
+                            attempt.candidate.joinGame(), targetDimension);
+                    try { output.writeBytes(respawns); } finally { respawns.release(); }
+                }
+                for (ByteBuf packet : queued) {
+                    ByteBuf mapped = host.keepAlives().body(frontend.alloc(), packet, false);
+                    if (mapped == null) continue;
+                    try {
+                        TransitionFrames.appendClientbound(output, frontend.alloc(), mapped,
+                                attempt.candidate.joinGame() == null ? null : attempt.candidate.joinGame().entityId(),
+                                clientEntityId);
+                    } finally {
+                        if (mapped != packet) mapped.release();
+                    }
+                }
+            });
+        } finally {
+            queued.forEach(ByteBuf::release);
+        }
+    }
+
+    private void failTransfer(TransferAttempt attempt, String reason) {
+        if (attempt.finished) return;
+        LOGGER.debug("Replacement backend failed for player {} to {}: {}; channel registered={}, active={}",
+                host.view().username(), attempt.target.address(), reason,
+                attempt.channel != null && attempt.channel.isRegistered(),
+                attempt.channel != null && attempt.channel.isActive());
+        attempt.finished = true;
+        attempt.failureReason = reason;
+        cancelCoordination(attempt);
+        if (attempt.channel != null) attempt.channel.close();
+        if (attempt.candidate != null) attempt.candidate.close();
+        if (attempt.frameState != null) attempt.frameState.close();
+        if (host.disconnecting()) return;
+        if (!attempt.pauseInProgress) resumeAfterFailedTransfer(attempt);
+    }
+
+    private void resumeAfterFailedTransfer(TransferAttempt attempt) {
+        if (attempt.paused && !attempt.detached && !host.closed() && !host.disconnecting()) {
+            attempt.oldRelay.resume().whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> {
+                if (failure != null) host.closeSession();
+                completeFailedTransfer(attempt);
+            }));
+        } else {
+            completeFailedTransfer(attempt);
+        }
+    }
+
+    private void completeFailedTransfer(TransferAttempt attempt) {
+        if (!attempt.detached && !host.closed() && !host.disconnecting()) {
+            try {
+                if (attempt.oldBackendBuffer != null) attempt.oldBackendBuffer.drainAndRemove();
+                if (attempt.clientBuffer != null) attempt.clientBuffer.drainAndRemove();
+            } catch (RuntimeException failure) {
+                host.closeSession();
+            }
+        }
+        if (attempt.cutoverDeadline != null) attempt.cutoverDeadline.cancel(false);
+        if (transfer == attempt) transfer = null;
+        attempt.result.complete(TransferResult.failed(attempt.failureReason));
+        if (attempt.detached) host.closeSession();
+    }
+
+    private static void cancelCoordination(TransferAttempt attempt) {
+        if (attempt.totalDeadline != null) attempt.totalDeadline.cancel(false);
+        CompletableFuture<?> work = attempt.coordination;
+        attempt.coordination = null;
+        // Cancellation may run arbitrary CompletionStage continuations, never on session I/O.
+        if (work != null) Thread.startVirtualThread(() -> work.cancel(false));
+    }
+    /** Session teardown: fails any queued or running transfer and releases the frame state. */
+    public void close() {
+        if (frameState != null) frameState.close();
+        PendingTransfer waiting = pendingTransfer;
+        if (waiting != null) {
+            pendingTransfer = null;
+            waiting.deadline.cancel(false);
+            waiting.result.complete(TransferResult.failed("player session closed during transfer"));
+        }
+        TransferAttempt activeTransfer = transfer;
+        if (activeTransfer != null) {
+            transfer = null;
+            activeTransfer.finished = true;
+            cancelCoordination(activeTransfer);
+            if (activeTransfer.cutoverDeadline != null) activeTransfer.cutoverDeadline.cancel(false);
+            if (activeTransfer.channel != null) activeTransfer.channel.close();
+            if (activeTransfer.candidate != null) activeTransfer.candidate.close();
+            if (activeTransfer.frameState != null) activeTransfer.frameState.close();
+            activeTransfer.result.complete(TransferResult.failed("player session closed during transfer"));
+        }
+    }
+}

proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferFrameBuffer.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.transfer;
 
 import dev.moonbridge.core.protocol.ProtocolProfile;
 import io.netty.buffer.ByteBuf;
@@ -9,7 +9,7 @@ import io.netty.util.ReferenceCountUtil;
 import java.util.ArrayDeque;
 
 /** Holds decoded frames on either side of the old session during backend cutover. */
-final class TransferFrameBuffer extends ChannelInboundHandlerAdapter {
+public final class TransferFrameBuffer extends ChannelInboundHandlerAdapter {
     // The decoder retains the three-byte length prefix on a maximum-sized protocol 5 frame.
     private static final int MAX_BYTES = ProtocolProfile.minecraft1710().maxFrameBytes() + 3;
     private static final int MAX_MESSAGES = 1024;
@@ -22,7 +22,7 @@ final class TransferFrameBuffer extends ChannelInboundHandlerAdapter {
     private boolean reading;
     private boolean discard;
 
-    TransferFrameBuffer(Runnable closeSession) {
+    public TransferFrameBuffer(Runnable closeSession) {
         this.closeSession = closeSession;
     }
 
@@ -31,7 +31,7 @@ final class TransferFrameBuffer extends ChannelInboundHandlerAdapter {
     }
 
     /** Keep reading bounded frames so a paused transfer also observes peer disconnects. */
-    void readUntilRemoved() {
+    public void readUntilRemoved() {
         ChannelHandlerContext ctx = context;
         if (ctx == null || !ctx.executor().inEventLoop()) {
             throw new IllegalStateException("transfer buffer must read on its channel event loop");
@@ -65,7 +65,7 @@ final class TransferFrameBuffer extends ChannelInboundHandlerAdapter {
     }
 
     /** Replays frames after this handler, preserving the already decoded frame boundary. */
-    void drainAndRemove() {
+    public void drainAndRemove() {
         var ctx = context;
         if (ctx == null || !ctx.executor().inEventLoop()) {
             throw new IllegalStateException("transfer buffer must drain on its channel event loop");
@@ -84,7 +84,7 @@ final class TransferFrameBuffer extends ChannelInboundHandlerAdapter {
     }
 
     /** After irreversible source detach, discard queued and future old-world frames. */
-    void discardFrames() {
+    public void discardFrames() {
         var ctx = context;
         if (ctx == null || !ctx.executor().inEventLoop()) {
             throw new IllegalStateException("transfer buffer must switch mode on its channel event loop");

proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferFrameHandler.java

@@ -1,4 +1,7 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.transfer;
+
+import dev.moonbridge.core.session.play.FrameTransformHandler;
+import dev.moonbridge.core.session.play.PlayObservation;
 
 import dev.moonbridge.core.protocol.Minecraft1710EntityIds;
 import dev.moonbridge.core.protocol.Minecraft1710PlayPackets;
@@ -12,8 +15,8 @@ import io.netty.channel.ChannelHandlerContext;
 import java.util.concurrent.CompletableFuture;
 
 /** Handles the few framed packets that cannot pass unchanged after a backend switch. */
-final class TransferFrameHandler extends FrameTransformHandler {
-    static final class State implements AutoCloseable {
+public final class TransferFrameHandler extends FrameTransformHandler {
+    public static final class State implements AutoCloseable {
         private final Channel frontend;
         private final Channel backend;
         private final PlayObservation observation;
@@ -78,7 +81,7 @@ final class TransferFrameHandler extends FrameTransformHandler {
     private final State state;
     private final boolean clientbound;
 
-    TransferFrameHandler(State state, boolean clientbound) {
+    public TransferFrameHandler(State state, boolean clientbound) {
         super(state.closeSession);
         this.state = state;
         this.clientbound = clientbound;

proxy-core/src/main/java/dev/moonbridge/core/session/transfer/TransferHost.java

@@ -0,0 +1,58 @@
+package dev.moonbridge.core.session.transfer;
+
+import dev.moonbridge.api.PlayerIdentity;
+import dev.moonbridge.api.PlayerView;
+import dev.moonbridge.core.backend.BackendCatalog;
+import dev.moonbridge.core.backend.BackendView;
+import dev.moonbridge.core.event.TransferPreparation;
+import dev.moonbridge.core.net.NetworkTransport;
+import dev.moonbridge.core.relay.RawRelay;
+import dev.moonbridge.core.session.play.KeepAliveBridge;
+import dev.moonbridge.core.session.play.PlayObservation;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.resolver.AddressResolverGroup;
+
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * What a {@link TransferCoordinator} needs from its session and the proxy. The session keeps ownership of
+ * the connection state; the coordinator only reads it and commits the two transitions below.
+ * All methods are called on the session's frontend event loop.
+ */
+public interface TransferHost {
+    Channel frontend();
+    Channel backend();
+    RawRelay.Link relay();
+    BackendView selected();
+    PlayerView view();
+    PlayerIdentity identity();
+    /** True once the player has been admitted to a backend (post login success). */
+    boolean published();
+    PlayObservation observation();
+    KeepAliveBridge.State keepAlives();
+    boolean closed();
+    boolean disconnecting();
+    void closeSession();
+
+    BackendCatalog catalog();
+    TransferPreparation selectPreparation();
+    Duration eventTimeout();
+    Duration cutoverTimeout();
+    NetworkTransport transport();
+    AddressResolverGroup<InetSocketAddress> resolver();
+
+    ByteBuf backendHandshake(String backendName, long backendEpoch);
+    void installTabCompletion(Channel backend);
+    /** Strips login-phase handlers from {@code channel} (after {@code beforeRemoval}) and installs lifecycle handling. */
+    CompletableFuture<Void> swapHandshakeCodecs(Channel channel, Runnable beforeRemoval);
+
+    /** The source backend is gone and no destination exists yet: the player has no current server. */
+    void sourceReleased();
+    /** The replacement backend now serves the player. */
+    void cutover(Channel backend, BackendView selected, PlayObservation observation, RawRelay.Link relay,
+                 Optional<String> previousServer);
+}

proxy-core/src/test/java/dev/moonbridge/core/session/ProxySessionListenerTest.java

@@ -1,5 +1,7 @@
 package dev.moonbridge.core.session;
 
+import dev.moonbridge.core.session.play.PlayObservation;
+
 import com.fasterxml.jackson.databind.ObjectMapper;
 import dev.moonbridge.api.PlacementDecision;
 import dev.moonbridge.api.AccessDecision;

proxy-core/src/test/java/dev/moonbridge/core/session/play/CommandRateLimiterTest.java

@@ -1,10 +1,10 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import org.junit.jupiter.api.Test;
 
-import static dev.moonbridge.core.session.CommandRateLimiter.Decision.ADMIT;
-import static dev.moonbridge.core.session.CommandRateLimiter.Decision.DENY;
-import static dev.moonbridge.core.session.CommandRateLimiter.Decision.DENY_WITH_NOTICE;
+import static dev.moonbridge.core.session.play.CommandRateLimiter.Decision.ADMIT;
+import static dev.moonbridge.core.session.play.CommandRateLimiter.Decision.DENY;
+import static dev.moonbridge.core.session.play.CommandRateLimiter.Decision.DENY_WITH_NOTICE;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 final class CommandRateLimiterTest {

proxy-core/src/test/java/dev/moonbridge/core/session/play/FrameTransformHandlerTest.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.DefaultByteBufHolder;

proxy-core/src/test/java/dev/moonbridge/core/session/play/KeepAliveBridgeTest.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import dev.moonbridge.core.relay.RawRelay;
 import io.netty.buffer.ByteBuf;

proxy-core/src/test/java/dev/moonbridge/core/session/play/PlayObservationTest.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import dev.moonbridge.core.protocol.Minecraft1710PlayPackets;
 import dev.moonbridge.core.protocol.ProtocolVarInt;

proxy-core/src/test/java/dev/moonbridge/core/session/play/TabCompletionBridgeTest.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import dev.moonbridge.core.protocol.Minecraft1710PlayPackets;
 import dev.moonbridge.core.protocol.MinecraftTabCompletion;

proxy-core/src/test/java/dev/moonbridge/core/session/play/TransitionBufferTest.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;

proxy-core/src/test/java/dev/moonbridge/core/session/play/TransitionFramesTest.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.play;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;

proxy-core/src/test/java/dev/moonbridge/core/session/transfer/TransferCandidateTest.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.transfer;
 
 import dev.moonbridge.core.protocol.Minecraft1710PlayPackets;
 import dev.moonbridge.core.protocol.ProtocolVarInt;

proxy-core/src/test/java/dev/moonbridge/core/session/transfer/TransferFrameBufferTest.java

@@ -1,4 +1,4 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.transfer;
 
 import dev.moonbridge.core.protocol.ProtocolProfile;
 import io.netty.buffer.ByteBuf;

proxy-core/src/test/java/dev/moonbridge/core/session/transfer/TransferFrameHandlerTest.java

@@ -1,4 +1,6 @@
-package dev.moonbridge.core.session;
+package dev.moonbridge.core.session.transfer;
+
+import dev.moonbridge.core.session.play.PlayObservation;
 
 import dev.moonbridge.core.protocol.Minecraft1710PlayPackets;
 import dev.moonbridge.core.protocol.ProtocolVarInt;