Add optional source-release transfer coordination events

docs/README.md

@@ -14,6 +14,7 @@
 | [架构与边界](architecture.md) | 模块职责、会话状态、线程模型、转服、目录所有权 |
 | [消息与命令补全](messages-and-completion.md) | 富文本、模板、异步参数补全及协议边界 |
 | [插件开发](plugins.md) | 生命周期、玩家/服务器视图、选服、命令与统一类型化事件 API |
+| [转服协调](transfer-coordination.md) | 登录前决策、可选源服释放、异步确认、期限与失败边界 |
 | [玩家权限与 LuckPerms](permissions.md) | 通用权限接口、提供者生命周期、命令鉴权和原生 LuckPerms 平台插件 |
 | [初次登录路由](initial-routing.md) | 显式入口、插件策略、超时、重试安全边界与验收契约 |
 | [玩家操作与部署完善设计](player-operations.md) | 消息、主动断开、服务器列表配置与验收边界 |

docs/transfer-coordination.md

@@ -0,0 +1,43 @@
+# Transfer coordination
+
+## Contract
+
+Transfer coordination is optional. Without `TransferPreparingEvent` subscribers, transfers use the existing parallel login and cutover path. Ordinary PLAY forwarding does not dispatch or inspect transfer events.
+
+After the destination TCP connection succeeds, before any destination Minecraft login bytes are written, the proxy calls the selected preparation listeners off its I/O thread. The source remains connected and usable. Listeners return `allow()`, `deny(reason)`, or `releaseSource(handler)`. Denial stops the transfer while preserving the source. All listeners are considered in registration order until denial; an allow cannot erase an earlier source-release requirement.
+
+Preparation should validate and record reversible intent. It must not take a final snapshot or clear live player state: the source can still process gameplay. A plugin that records intent must reconcile abandoned intent independently; notification delivery is not a persistence guarantee.
+
+If any listener requests source release, the proxy stops forwarding new source-world operations, drains already accepted network writes, detaches the source relay, and closes the source connection. It then invokes only the callbacks attached to the selected source-release decisions, in registration order, off the I/O thread. `SourceReleasedEvent` is callback context, not an independently subscribable notification. Socket closure does not mean the backend has processed logout or committed a save. Each required callback defines and confirms its own durable boundary through successful asynchronous completion.
+
+Only after all required callbacks succeed does the proxy send the destination Handshake and LoginStart. It checks that the attempt, frontend connection, destination connection and destination registration generation are still current. Source-world packets received during the gap are discarded, never queued for replay into the destination. Existing Forge negotiation continues to permit control traffic and reject gameplay until its world boundary is ready.
+
+## Identity and lifetime
+
+`TransferContext` is an immutable snapshot containing a unique transfer ID, exact player/frontend connection identity, source and destination server snapshots, catalog registration generations, authenticated backend instance epochs (zero for static backends without an instance binding), and a total deadline. It exposes no channels, buffers, mutable sessions, database concepts or player inventory logic.
+
+The listener cohort is selected once per attempt. Every required source-release callback remains bound to its original registration and plugin owner. Revocation, plugin shutdown, null results, exception, timeout or overload fails the transfer; it cannot silently shrink the required cohort. Late completions cannot resume a closed or replaced attempt.
+
+Callbacks use a separate bounded dispatcher: two workers, 128 queued tasks and at most 128 pending events. Callback stages do not block I/O threads or globally serialize unrelated transfers. No subscribers means no transfer event, event future or event timer allocation. The dispatcher is created only when subscriptions exist. A slow plugin can cause its transfer to fail; it cannot create an unbounded queue.
+
+## Deadlines and failures
+
+Preparation and source-release confirmation each use the configured event deadline, including queue wait. Destination login gets its own 15-second budget after login starts for coordinated transfers. TCP connection retains its five-second budget. The existing cutover and Forge negotiation budgets remain. A total deadline bounds the entire coordinated attempt.
+
+| Failure point | Outcome |
+| --- | --- |
+| TCP failure, preparation denial/error/timeout | Close candidate; source continues |
+| Pause failure or partial detach | Preserve source only if its relay can be safely restored; otherwise disconnect |
+| Source released, callback error/timeout or destination failure | Disconnect frontend; never fall back to the closed source |
+| Destination registration changed | Fail this attempt; never authenticate against a replacement instance implicitly |
+| Frontend disconnect or proxy shutdown | Close candidate and release local resources; ignore late results |
+
+This opt-in sequence sacrifices the normal destination-login-before-source-close fallback window. TCP reachability does not prove whitelist, authentication, mod negotiation or gameplay readiness. `TransferResult.NETWORK_READY` retains its existing meaning; it does not certify persistence by itself.
+
+## Acceptance
+
+Correctness tests must establish source close before confirmation, confirmation before destination login, denial preserving the source, failure after release disconnecting, old source EOF not closing the waiting frontend, cancellation and stale callbacks, exact backend generation checks, and no old-world gameplay replay. Existing vanilla and Forge transfers must remain covered.
+
+Measure ordinary PLAY forwarding against main `876de1a056a79b807a03f3877cbe6414b60e86dc`, with initial and post-transfer sessions, windows 1/16 and five repetitions. Also measure transfer throughput and p50/p99 latency at 0/1/4 participants and 1/16/64 concurrent players. Record environment, errors, allocation and retained buffering where measurable. An initial 3% default-path regression budget is a decision threshold, not a result; if measurement noise exceeds it, classify the comparison as inconclusive and improve the experiment. Passing functional tests does not establish performance superiority.
+
+PlayerDataSQL integration is a separate plugin concern. These events do not implement SQL ownership, backend final-save confirmation, authoritative data loading, mod synchronization or crash recovery.

proxy-core/src/main/java/dev/moonbridge/core/event/EventDispatcher.java

@@ -9,4 +9,12 @@ public interface EventDispatcher {
     boolean hasSubscribers(Class<?> eventType);
 
     <R> CompletionStage<R> dispatch(Event<R> event);
+
+    /**
+     * Pins the currently enabled transfer-preparation listener cohort. Returns {@code null} when
+     * no transfer-preparation listener was registered at enable time.
+     */
+    default TransferPreparation selectTransferPreparation() {
+        return null;
+    }
 }

proxy-core/src/main/java/dev/moonbridge/core/event/PreparedTransfer.java

@@ -0,0 +1,13 @@
+package dev.moonbridge.core.event;
+
+import dev.moonbridge.api.event.SourceReleasedEvent;
+import java.util.concurrent.CompletionStage;
+
+/** Immutable result of transfer preparation, retained by the core for the selected transfer. */
+public interface PreparedTransfer {
+    /** Whether one or more selected listeners require the source-release phase. */
+    boolean requiresSourceRelease();
+
+    /** Runs required callbacks in registration order after the exact source socket closes. */
+    CompletionStage<Void> sourceReleased(SourceReleasedEvent event);
+}

proxy-core/src/main/java/dev/moonbridge/core/event/TransferPreparation.java

@@ -0,0 +1,9 @@
+package dev.moonbridge.core.event;
+
+import dev.moonbridge.api.event.TransferPreparingEvent;
+import java.util.concurrent.CompletionStage;
+
+/** Host-selected, owner-pinned transfer preparation cohort for one transfer. */
+public interface TransferPreparation {
+    CompletionStage<PreparedTransfer> prepare(TransferPreparingEvent event);
+}

proxy-core/src/main/java/dev/moonbridge/core/plugin/PluginHost.java

@@ -11,6 +11,9 @@ import dev.moonbridge.api.event.Events;
 import dev.moonbridge.api.event.PlayerAdmissionEvent;
 import dev.moonbridge.api.event.PlayerDisconnectedEvent;
 import dev.moonbridge.api.event.ServerConnectedEvent;
+import dev.moonbridge.api.event.SourceReleasedEvent;
+import dev.moonbridge.api.event.TransferDecision;
+import dev.moonbridge.api.event.TransferPreparingEvent;
 import dev.moonbridge.messaging.Endpoint;
 import dev.moonbridge.messaging.Message;
 import dev.moonbridge.messaging.MessageKind;
@@ -50,7 +53,9 @@ import dev.moonbridge.core.backend.BackendRegistration;
 import dev.moonbridge.core.backend.BackendView;
 import dev.moonbridge.core.control.BackendChannelTransport;
 import dev.moonbridge.core.event.EventDispatcher;
+import dev.moonbridge.core.event.PreparedTransfer;
 import dev.moonbridge.core.permission.PermissionService;
+import dev.moonbridge.core.event.TransferPreparation;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -94,6 +99,7 @@ import java.util.concurrent.atomic.AtomicReference;
 import java.util.UUID;
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import java.util.jar.JarFile;
 
 /** Owns plugin lifecycles and exposes only the proxy services in the plugin API. */
@@ -102,6 +108,14 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
     private static final AtomicLong NEXT_HOST_GENERATION = new AtomicLong();
     private static final int MAX_PENDING_COMMAND_COMPLETIONS = 128;
     private static final int MAX_PENDING_COMMAND_EXECUTIONS = 128;
+    private static final int MAX_PENDING_TRANSFER_EVENTS = 128;
+    private static final PreparedTransfer ALLOWED_TRANSFER = new PreparedTransfer() {
+        @Override public boolean requiresSourceRelease() { return false; }
+        @Override public CompletionStage<Void> sourceReleased(SourceReleasedEvent event) {
+            Objects.requireNonNull(event, "event");
+            return CompletableFuture.completedFuture(null);
+        }
+    };
     private static final int MAX_COMPLETION_RESULTS = 100;
     // Leave room for bounded protocol encoding overhead beyond suggestion UTF-8 bodies.
     private static final int MAX_COMPLETION_BYTES = 32_000;
@@ -114,6 +128,7 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
     private final Players players;
     private final UUID proxyEpoch;
     private final Duration placementTimeout;
+    private final Duration eventTimeout;
     private final PermissionService permissionService;
     private final Duration shutdownTimeout;
     private final ThreadPoolExecutor callbacks;
@@ -128,6 +143,7 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
     private final ScheduledThreadPoolExecutor timer;
     private final AsyncEventDispatcher admissionEvents;
     private final AsyncEventDispatcher notificationEvents;
+    private final AsyncEventDispatcher.EventPolicy<TransferDecision> transferPolicy;
     private final AsyncEventDispatcher.EventPolicy<AccessDecision> admissionPolicy;
     private final AsyncEventDispatcher.EventPolicy<Void> notificationPolicy;
     private final AtomicLong notificationDrops = new AtomicLong();
@@ -140,6 +156,7 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
     private ScheduledFuture<?> notificationDropReporter;
     private InitialPlacementHandler placementHandler;
     private volatile Map<Class<?>, List<EventRegistration<?, ?>>> eventListeners = Map.of();
+    private volatile AsyncEventDispatcher transferEvents;
 
     public PluginHost(BackendCatalog catalog, Players players, Duration placementTimeout) {
         this(catalog, players, placementTimeout, Duration.ofSeconds(5));
@@ -193,6 +210,7 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
         this.proxyEpoch = Objects.requireNonNull(proxyEpoch, "proxyEpoch");
         this.placementTimeout = Objects.requireNonNull(placementTimeout, "placementTimeout");
         Objects.requireNonNull(eventTimeout, "eventTimeout");
+        this.eventTimeout = eventTimeout;
         this.permissionService = new PermissionService(eventTimeout);
         this.shutdownTimeout = Objects.requireNonNull(shutdownTimeout, "shutdownTimeout");
         if (placementTimeout.isZero() || placementTimeout.isNegative()) {
@@ -222,6 +240,11 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
         this.notificationPolicy = new AsyncEventDispatcher.EventPolicy<>(() -> null, ignored -> true,
                 ignored -> false, true, (event, failure) -> LOGGER.warn("Plugin event listener failed for {}",
                 event.getClass().getName(), failure), 129, true);
+        this.transferPolicy = new AsyncEventDispatcher.EventPolicy<>(TransferDecision::allow,
+                decision -> decision instanceof TransferDecision.Allowed
+                        || decision instanceof TransferDecision.Denied,
+                decision -> decision instanceof TransferDecision.Denied, false, (event, failure) -> { },
+                MAX_PENDING_TRANSFER_EVENTS, false);
         this.admissionEvents = new AsyncEventDispatcher(eventTimeout, accessThreads, accessQueueCapacity,
                 timer, namedThreads("moonbridge-plugin-admission"));
         this.notificationEvents = new AsyncEventDispatcher(eventTimeout, 1, 128,
@@ -363,6 +386,11 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
             }
             if (selectedPermissions != null) permissionService.configure(selectedPermissions);
             freezeEventRegistrations();
+            if (eventListeners.getOrDefault(TransferPreparingEvent.class, List.of()).stream()
+                    .anyMatch(EventRegistration::isActive)) {
+                transferEvents = new AsyncEventDispatcher(eventTimeout, 2, 128,
+                        timer, namedThreads("moonbridge-plugin-transfer"));
+            }
             state = State.ENABLED;
             if (eventListeners.containsKey(ServerConnectedEvent.class)
                     || eventListeners.containsKey(PlayerDisconnectedEvent.class)) {
@@ -401,10 +429,127 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
             });
             return delivery;
         }
+        if (event instanceof TransferPreparingEvent) {
+            return CompletableFuture.failedFuture(new IllegalArgumentException(
+                    "Transfer preparation must use a pinned transfer cohort"));
+        }
         return CompletableFuture.failedFuture(new IllegalArgumentException(
                 "Unsupported event type: " + event.getClass().getName()));
     }
 
+    @Override
+    public TransferPreparation selectTransferPreparation() {
+        if (state != State.ENABLED) return null;
+        List<EventRegistration<?, ?>> registered = eventListeners.getOrDefault(TransferPreparingEvent.class, List.of());
+        AsyncEventDispatcher dispatcher = transferEvents;
+        if (registered.isEmpty() || dispatcher == null) return null;
+        List<EventRegistration<?, ?>> cohort = registered.stream().filter(EventRegistration::isActive).toList();
+        if (cohort.isEmpty()) return null;
+        return event -> prepareTransfer(dispatcher, cohort, event);
+    }
+
+    @SuppressWarnings({"unchecked", "rawtypes"})
+    private CompletionStage<PreparedTransfer> prepareTransfer(AsyncEventDispatcher dispatcher,
+            List<EventRegistration<?, ?>> cohort, TransferPreparingEvent event) {
+        Objects.requireNonNull(event, "event");
+        List<PinnedReleaseCallback> callbacks = new ArrayList<>();
+        List<AsyncEventDispatcher.EventHandler<TransferDecision>> handlers = new ArrayList<>(cohort.size());
+        for (EventRegistration<?, ?> raw : cohort) {
+            EventRegistration<TransferPreparingEvent, TransferDecision> registration = (EventRegistration) raw;
+            handlers.add(new AsyncEventDispatcher.EventHandler<>() {
+                @Override public boolean active() { return true; }
+
+                @Override public CompletionStage<TransferDecision> handle(Event<?> ignored) {
+                    registration.requireActive();
+                    CompletionStage<TransferDecision> stage = Objects.requireNonNull(
+                            registration.handle(event), "transfer preparation stage");
+                    return mapCancellable(stage, decision -> {
+                        registration.requireActive();
+                        Objects.requireNonNull(decision, "transfer decision");
+                        if (decision instanceof TransferDecision.Denied) return decision;
+                        if (decision instanceof TransferDecision.ReleaseSource release) {
+                            callbacks.add(new PinnedReleaseCallback(registration, release.handler()));
+                        } else if (!(decision instanceof TransferDecision.Allowed)) {
+                            throw new IllegalStateException("Unsupported transfer decision");
+                        }
+                        return TransferDecision.allow();
+                    });
+                }
+            });
+        }
+        CompletionStage<TransferDecision> dispatched = dispatcher.dispatch(event, handlers, transferPolicy);
+        return mapCancellable(dispatched, decision -> {
+            if (decision instanceof TransferDecision.Denied) {
+                LOGGER.info("Transfer preparation denied for transfer {}", event.context().transferId());
+                throw new IllegalStateException("transfer preparation denied");
+            }
+            List<PinnedReleaseCallback> pinned = List.copyOf(callbacks);
+            if (pinned.isEmpty()) return ALLOWED_TRANSFER;
+            return new PreparedTransfer() {
+                private final AtomicBoolean releaseStarted = new AtomicBoolean();
+
+                @Override public boolean requiresSourceRelease() { return !pinned.isEmpty(); }
+
+                @Override public CompletionStage<Void> sourceReleased(SourceReleasedEvent released) {
+                    Objects.requireNonNull(released, "released");
+                    if (!event.context().equals(released.context())) {
+                        return CompletableFuture.failedFuture(
+                                new IllegalArgumentException("source release context does not match transfer"));
+                    }
+                    if (pinned.isEmpty()) return CompletableFuture.completedFuture(null);
+                    if (!releaseStarted.compareAndSet(false, true)) {
+                        return CompletableFuture.failedFuture(
+                                new IllegalStateException("source release callbacks already started"));
+                    }
+                    List<AsyncEventDispatcher.EventHandler<TransferDecision>> releaseHandlers = pinned.stream()
+                            .<AsyncEventDispatcher.EventHandler<TransferDecision>>map(callback ->
+                                    new AsyncEventDispatcher.EventHandler<>() {
+                                        @Override public boolean active() { return true; }
+                                        @Override public CompletionStage<TransferDecision> handle(Event<?> ignored) {
+                                            return mapCancellable(callback.invoke(released),
+                                                    nothing -> TransferDecision.allow());
+                                        }
+                                    }).toList();
+                    CompletionStage<TransferDecision> release = dispatcher.dispatch(event, releaseHandlers,
+                            transferPolicy);
+                    return mapCancellable(release, ignored -> {
+                        pinned.forEach(PinnedReleaseCallback::requireActive);
+                        return null;
+                    });
+                }
+            };
+        });
+    }
+
+    private static <A, B> CompletableFuture<B> mapCancellable(
+            CompletionStage<A> stage, Function<? super A, ? extends B> mapper) {
+        CompletableFuture<A> upstream = stage.toCompletableFuture();
+        Object completionLock = new Object();
+        CompletableFuture<B> mapped = new CompletableFuture<>() {
+            @Override public boolean cancel(boolean mayInterruptIfRunning) {
+                synchronized (completionLock) {
+                    boolean cancelled = super.cancel(mayInterruptIfRunning);
+                    if (cancelled) upstream.cancel(mayInterruptIfRunning);
+                    return cancelled;
+                }
+            }
+        };
+        upstream.whenComplete((value, failure) -> {
+            synchronized (completionLock) {
+                if (mapped.isDone()) return;
+                if (failure != null) mapped.completeExceptionally(failure);
+                else {
+                    try {
+                        mapped.complete(mapper.apply(value));
+                    } catch (Throwable mappingFailure) {
+                        mapped.completeExceptionally(mappingFailure);
+                    }
+                }
+            }
+        });
+        return mapped;
+    }
+
     @SuppressWarnings({"unchecked", "rawtypes"})
     private <R> CompletionStage<R> dispatchWithPolicy(AsyncEventDispatcher dispatcher, Event<R> event,
                                                        List<EventRegistration<?, ?>> registrations,
@@ -920,6 +1065,7 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
         // chain can skip the revoked subscriptions and incorrectly finish with allow().
         admissionEvents.close();
         notificationEvents.close();
+        if (transferEvents != null) transferEvents.close();
         localMessaging.close();
         if (notificationDropReporter != null) notificationDropReporter.cancel(false);
         logNotificationDrops();
@@ -1019,7 +1165,8 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
         Objects.requireNonNull(eventType, "eventType");
         Objects.requireNonNull(listener, "listener");
         if (eventType != ConnectionAdmissionEvent.class && eventType != PlayerAdmissionEvent.class
-                && eventType != ServerConnectedEvent.class && eventType != PlayerDisconnectedEvent.class)
+                && eventType != ServerConnectedEvent.class && eventType != PlayerDisconnectedEvent.class
+                && eventType != TransferPreparingEvent.class)
             throw new IllegalArgumentException("Unsupported event type: " + eventType.getName());
         requireEventRegistrationState();
         synchronized (this) {
@@ -1167,9 +1314,37 @@ public final class PluginHost implements AutoCloseable, EventDispatcher {
             return listener.onEvent(eventType.cast(event));
         }
         private boolean isActive() { return active.get() && context.active; }
+        private void requireActive() {
+            if (!isActive()) throw new IllegalStateException("Transfer listener owner is no longer active");
+        }
         @Override public void close() { active.set(false); }
     }
 
+    private final class PinnedReleaseCallback {
+        private final EventRegistration<TransferPreparingEvent, TransferDecision> registration;
+        private final dev.moonbridge.api.event.SourceReleasedHandler handler;
+
+        private PinnedReleaseCallback(EventRegistration<TransferPreparingEvent, TransferDecision> registration,
+                                      dev.moonbridge.api.event.SourceReleasedHandler handler) {
+            this.registration = registration;
+            this.handler = handler;
+        }
+
+        private CompletionStage<Void> invoke(SourceReleasedEvent event) {
+            registration.requireActive();
+            CompletionStage<Void> stage = Objects.requireNonNull(
+                    handler.onSourceReleased(event), "source release callback stage");
+            return mapCancellable(stage, ignored -> {
+                requireActive();
+                return null;
+            });
+        }
+
+        private void requireActive() {
+            registration.requireActive();
+        }
+    }
+
     private static final class LoadedPlugin {
         private final Plugin plugin;
         private final BackendOwner owner;

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

@@ -297,9 +297,15 @@ public final class ProxySessionListener implements Players {
     }
 
     boolean hasSubscribers(Class<?> eventType) { return events != null && events.hasSubscribers(eventType); }
+    dev.moonbridge.core.event.TransferPreparation selectTransferPreparation() {
+        return events == null ? null : events.selectTransferPreparation();
+    }
     <R> CompletionStage<R> dispatchEvent(Event<R> event) { return events.dispatch(event); }
-    void serverConnected(PlayerView player, Optional<String> previousServer) {
+    void playerContextChanged(PlayerView player) {
         if (permissions != null) permissions.update(player);
+    }
+    void serverConnected(PlayerView player, Optional<String> previousServer) {
+        playerContextChanged(player);
         if (hasSubscribers(ServerConnectedEvent.class)) events.dispatch(new ServerConnectedEvent(player, previousServer));
     }
     void playerDisconnected(PlayerView player) {

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

@@ -2,6 +2,12 @@ package dev.moonbridge.core.session;
 
 import dev.moonbridge.api.PlacementDecision;
 import dev.moonbridge.api.AccessDecision;
+import dev.moonbridge.api.ServerView;
+import dev.moonbridge.api.event.TransferContext;
+import dev.moonbridge.api.event.TransferPreparingEvent;
+import dev.moonbridge.api.event.SourceReleasedEvent;
+import dev.moonbridge.core.event.TransferPreparation;
+import dev.moonbridge.core.event.PreparedTransfer;
 import dev.moonbridge.api.event.PlayerAdmissionEvent;
 import dev.moonbridge.api.PlayerIdentity;
 import dev.moonbridge.api.PlayerView;
@@ -49,6 +55,7 @@ import java.net.URI;
 import java.security.GeneralSecurityException;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.ArrayDeque;
@@ -969,9 +976,14 @@ final class Session extends ChannelInboundHandlerAdapter {
     }
 
     private CompletableFuture<Void> removeHandshakeCodecs(Channel channel) {
+        return removeHandshakeCodecs(channel, () -> { });
+    }
+
+    private CompletableFuture<Void> removeHandshakeCodecs(Channel channel, Runnable beforeRemoval) {
         CompletableFuture<Void> removed = new CompletableFuture<>();
         channel.eventLoop().execute(() -> {
             try {
+                beforeRemoval.run();
                 ChannelPipeline pipeline = channel.pipeline();
                 if (pipeline.get("initial-session") != null) pipeline.remove("initial-session");
                 if (pipeline.get("transfer-candidate") != null) pipeline.remove("transfer-candidate");
@@ -1063,13 +1075,25 @@ final class Session extends ChannelInboundHandlerAdapter {
             return;
         }
         TransferAttempt attempt = new TransferAttempt(target, result, relay);
+        attempt.preparation = owner.selectTransferPreparation();
+        if (attempt.preparation != null) {
+            Duration totalBudget = owner.eventTimeout().multipliedBy(2).plusSeconds(20)
+                    .plus(owner.transferCutoverTimeout()).plus(FORGE_TRANSFER_HANDSHAKE_TIMEOUT);
+            attempt.context = new TransferContext(UUID.randomUUID(), identity, serverView(selected), serverView(target),
+                    selected.handle().generation(), target.handle().generation(),
+                    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 {}",
                 view.username(), target.address());
         attempt.candidate = new TransferCandidate(identity.playerId(), 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 = new Bootstrap().group(frontend.eventLoop()).channel(NioSocketChannel.class)
                     .resolver(owner.backendResolver())
@@ -1097,13 +1121,143 @@ final class Session extends ChannelInboundHandlerAdapter {
                 }
                 LOGGER.debug("Replacement backend TCP connection established for player {} to {}",
                         view.username(), attempt.target.address());
-                writeBackendLogin(connect.channel());
+                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 && !closed.get() && !disconnecting;
+    }
+
+    private boolean targetStillCurrent(TransferAttempt attempt) {
+        BackendView current = owner.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 {}", 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 = backend;
+        try {
+            attempt.clientBuffer = installTransferBuffer(frontend, "transfer-client-buffer");
+            attempt.oldBackendBuffer = installTransferBuffer(source, "transfer-old-backend-buffer");
+        } catch (RuntimeException failure) {
+            failTransfer(attempt, "could not stop source gameplay for transfer");
+            closePair();
+            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, pauseFailure) -> frontend.eventLoop().execute(() -> {
+            attempt.pauseInProgress = false;
+            if (pauseFailure != null) {
+                failTransfer(attempt, "could not drain source relay");
+                closePair();
+                return;
+            }
+            attempt.paused = true;
+            if (!currentAttempt(attempt)) {
+                if (!closed.get() && !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) closePair();
+                    return;
+                }
+                attempt.detached = true;
+                if (!currentAttempt(attempt) || !targetStillCurrent(attempt)) {
+                    failTransfer(attempt, "replacement backend closed during source release");
+                    closePair();
+                    return;
+                }
+                // Disarm exact old-channel EOF before closing it; no destination exists yet.
+                attempt.sourceReleased = true;
+                attempt.clientBuffer.discardFrames();
+                attempt.oldBackendBuffer.discardFrames();
+                backend = null;
+                relay = null;
+                view = new PlayerView(identity, view.username(), Optional.empty());
+                owner.playerContextChanged(view);
+                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.sourceReleased(new SourceReleasedEvent(attempt.context));
+            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 {}", 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");
+        }
+    }
+
     private void tryStartPendingTransfer() {
         PendingTransfer waiting = pendingTransfer;
         if (disconnecting || waiting == null || relay == null || !playObservation.ready().isDone()) return;
@@ -1113,6 +1267,7 @@ final class Session extends ChannelInboundHandlerAdapter {
     }
 
     private void writeBackendLogin(Channel channel) {
+        transfer.candidate.loginStarted();
         ByteBuf handshakeBody = encodeBackendHandshake(
                 transfer.target.handle().id().value(), transfer.target.owner().instanceGeneration());
         ByteBuf loginBody = new LoginStart(view.username()).encode(frontend.alloc(), ProtocolProfile.minecraft1710());
@@ -1186,11 +1341,23 @@ final class Session extends ChannelInboundHandlerAdapter {
 
     private void candidateReady(TransferAttempt attempt) {
         if (transfer != attempt || attempt.finished || closed.get() || 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"));
             closePair();
         }, owner.transferCutoverTimeout().toNanos(), TimeUnit.NANOSECONDS);
+        if (attempt.sourceReleased) {
+            if (clientHasPartialFrame()) {
+                failTransfer(attempt, "client packet was incomplete at replacement cutover");
+                return;
+            }
+            activateCandidate(attempt);
+            return;
+        }
         try {
             attempt.clientBuffer = installTransferBuffer(frontend, "transfer-client-buffer");
             attempt.oldBackendBuffer = installTransferBuffer(backend, "transfer-old-backend-buffer");
@@ -1265,14 +1432,9 @@ final class Session extends ChannelInboundHandlerAdapter {
     }
 
     private void activateCandidate(TransferAttempt attempt) {
-        try {
-            attempt.candidate.handOff();
-        } catch (RuntimeException invalid) {
-            failTransfer(attempt, "replacement backend was not ready");
-            closePair();
-            return;
-        }
-        removeHandshakeCodecs(attempt.channel).whenComplete((ignored, failure) ->
+        // 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.
+        removeHandshakeCodecs(attempt.channel, attempt.candidate::handOff).whenComplete((ignored, failure) ->
                 frontend.eventLoop().execute(() -> {
                     if (failure != null || closed.get() || disconnecting || !attempt.channel.isActive()) {
                         if (disconnecting) return;
@@ -1306,7 +1468,8 @@ final class Session extends ChannelInboundHandlerAdapter {
             return;
         }
         next.ready().whenComplete((ignored, failure) -> frontend.eventLoop().execute(() -> {
-            if (failure != null || closed.get() || disconnecting) {
+            if (failure != null || closed.get() || disconnecting
+                    || (attempt.preparation != null && !targetStillCurrent(attempt))) {
                 if (disconnecting) return;
                 failTransfer(attempt, "replacement relay was not ready");
                 closePair();
@@ -1322,14 +1485,16 @@ final class Session extends ChannelInboundHandlerAdapter {
                 return;
             }
             frontend.writeAndFlush(opening).addListener(write -> frontend.eventLoop().execute(() -> {
-                if (!write.isSuccess() || closed.get() || disconnecting || !attempt.channel.isActive()) {
+                if (!write.isSuccess() || closed.get() || disconnecting || !attempt.channel.isActive()
+                        || (attempt.preparation != null && !targetStillCurrent(attempt))) {
                     if (disconnecting) return;
                     failTransfer(attempt, "could not send replacement world transition");
                     closePair();
                     return;
                 }
                 Channel oldBackend = backend;
-                Optional<String> previousServer = view.currentServer();
+                Optional<String> previousServer = attempt.sourceReleased
+                        ? Optional.of(attempt.context.source().name()) : view.currentServer();
                 PlayObservation oldObservation = playObservation;
                 TransferFrameHandler.State oldFrameState = frameState;
                 backend = attempt.channel;
@@ -1339,7 +1504,7 @@ final class Session extends ChannelInboundHandlerAdapter {
                 frameState = attempt.frameState;
                 view = new PlayerView(identity, view.username(), selected.handle().id().value());
                 owner.serverConnected(view, previousServer);
-                oldBackend.close();
+                if (oldBackend != null) oldBackend.close();
                 oldObservation.close();
                 if (oldFrameState != null) oldFrameState.close();
                 try {
@@ -1376,6 +1541,7 @@ final class Session extends ChannelInboundHandlerAdapter {
         transfer = null;
         attempt.finished = true;
         attempt.cutoverDeadline.cancel(false);
+        cancelCoordination(attempt);
         attempt.result.complete(TransferResult.of(TransferStatus.NETWORK_READY));
     }
 
@@ -1444,6 +1610,7 @@ final class Session extends ChannelInboundHandlerAdapter {
                 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();
@@ -1474,6 +1641,15 @@ final class Session extends ChannelInboundHandlerAdapter {
         if (attempt.cutoverDeadline != null) attempt.cutoverDeadline.cancel(false);
         if (transfer == attempt) transfer = null;
         attempt.result.complete(TransferResult.failed(attempt.failureReason));
+        if (attempt.detached) closePair();
+    }
+
+    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));
     }
 
     void closePair() {
@@ -1507,6 +1683,7 @@ final class Session extends ChannelInboundHandlerAdapter {
             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();
@@ -1587,6 +1764,11 @@ final class Session extends ChannelInboundHandlerAdapter {
         private boolean finished;
         private String failureReason;
         private ScheduledFuture<?> cutoverDeadline;
+        private TransferPreparation preparation;
+        private TransferContext context;
+        private CompletableFuture<?> coordination;
+        private ScheduledFuture<?> totalDeadline;
+        private boolean sourceReleased;
 
         private TransferAttempt(BackendView target,
                                 CompletableFuture<TransferResult> result, RawRelay.Link oldRelay) {

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

@@ -28,6 +28,9 @@ final class TransferCandidate extends ChannelInboundHandlerAdapter {
     private final UUID expectedId;
     private final String username;
     private final Listener listener;
+    private final boolean deferLogin;
+    private ChannelHandlerContext context;
+    private boolean loginStarted;
     private final PlayObservation observation = new PlayObservation();
     private final ArrayDeque<ByteBuf> queued = new ArrayDeque<>();
     private Minecraft1710PlayPackets.JoinGame joinGame;
@@ -36,13 +39,30 @@ final class TransferCandidate extends ChannelInboundHandlerAdapter {
     private int queuedBytes;
 
     TransferCandidate(UUID expectedId, String username, Listener listener) {
+        this(expectedId, username, listener, false);
+    }
+
+    TransferCandidate(UUID expectedId, String username, Listener listener, boolean deferLogin) {
         this.expectedId = expectedId;
         this.username = username;
         this.listener = listener;
+        this.deferLogin = deferLogin;
     }
 
     @Override public void handlerAdded(ChannelHandlerContext ctx) {
-        deadline = ctx.executor().schedule(() -> fail(ctx, "backend login timed out"), 15, TimeUnit.SECONDS);
+        context = ctx;
+        if (!deferLogin) loginStarted();
+    }
+
+    /** Coordinated waits must not consume the destination's login budget. */
+    void loginStarted() {
+        if (loginStarted) return;
+        if (context == null || !context.executor().inEventLoop()) {
+            throw new IllegalStateException("candidate login must start on its event loop");
+        }
+        loginStarted = true;
+        deadline = context.executor().schedule(() -> fail(context, "backend login timed out"),
+                15, TimeUnit.SECONDS);
     }
 
     @Override public void channelRead(ChannelHandlerContext ctx, Object message) {
@@ -58,6 +78,10 @@ final class TransferCandidate extends ChannelInboundHandlerAdapter {
                 throw new IllegalArgumentException("invalid backend transfer frame");
             }
             if (state == State.FAILED || state == State.HANDED_OFF) return;
+            if (!loginStarted) {
+                fail(ctx, "backend sent data before login started");
+                return;
+            }
             if (state == State.LOGIN) {
                 int id = ProtocolVarInt.read(packet.duplicate());
                 if (id != 2) {

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

@@ -20,6 +20,7 @@ final class TransferFrameBuffer extends ChannelInboundHandlerAdapter {
     private int bytes;
     private boolean draining;
     private boolean reading;
+    private boolean discard;
 
     TransferFrameBuffer(Runnable closeSession) {
         this.closeSession = closeSession;
@@ -49,6 +50,10 @@ final class TransferFrameBuffer extends ChannelInboundHandlerAdapter {
             closeSession.run();
             return;
         }
+        if (discard) {
+            frame.release();
+            return;
+        }
         int size = frame.readableBytes();
         if (frames.size() >= MAX_MESSAGES || size > MAX_BYTES - bytes) {
             frame.release();
@@ -78,6 +83,16 @@ final class TransferFrameBuffer extends ChannelInboundHandlerAdapter {
         }
     }
 
+    /** After irreversible source detach, discard queued and future old-world frames. */
+    void discardFrames() {
+        var ctx = context;
+        if (ctx == null || !ctx.executor().inEventLoop()) {
+            throw new IllegalStateException("transfer buffer must switch mode on its channel event loop");
+        }
+        discard = true;
+        releaseFrames();
+    }
+
     @Override public void handlerRemoved(ChannelHandlerContext ctx) {
         releaseFrames();
     }

proxy-core/src/test/java/dev/moonbridge/core/plugin/PluginHostTest.java

@@ -11,6 +11,10 @@ import dev.moonbridge.api.event.Events;
 import dev.moonbridge.api.event.PlayerAdmissionEvent;
 import dev.moonbridge.api.event.PlayerDisconnectedEvent;
 import dev.moonbridge.api.event.ServerConnectedEvent;
+import dev.moonbridge.api.event.SourceReleasedEvent;
+import dev.moonbridge.api.event.TransferContext;
+import dev.moonbridge.api.event.TransferDecision;
+import dev.moonbridge.api.event.TransferPreparingEvent;
 
 import dev.moonbridge.api.CommandInvocation;
 import dev.moonbridge.api.PlacementDecision;
@@ -32,16 +36,19 @@ import dev.moonbridge.core.backend.BackendOwner;
 import dev.moonbridge.core.backend.BackendRegistration;
 import dev.moonbridge.core.backend.BackendView;
 import dev.moonbridge.core.backend.InMemoryBackendCatalog;
+import dev.moonbridge.core.event.PreparedTransfer;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.URI;
+import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -127,6 +134,214 @@ class PluginHostTest {
         }
     }
 
+    @Test
+    void transferPreparationPinsListenersAndRunsReleaseCallbacksInRegistrationOrder() throws Exception {
+        var order = new java.util.concurrent.CopyOnWriteArrayList<String>();
+        TransferContext context = transferContext();
+        var first = new Plugin() {
+            @Override public void onLoad(PluginContext pluginContext) {
+                pluginContext.events().subscribe(TransferPreparingEvent.class, event -> {
+                    order.add("prepare-first");
+                    return CompletableFuture.completedFuture(TransferDecision.releaseSource(released -> {
+                        assertEquals(context, released.context());
+                        order.add("release-first");
+                        return CompletableFuture.completedFuture(null);
+                    }));
+                });
+            }
+        };
+        var second = new Plugin() {
+            @Override public void onLoad(PluginContext pluginContext) {
+                pluginContext.events().subscribe(TransferPreparingEvent.class, event -> {
+                    order.add("prepare-second");
+                    return CompletableFuture.completedFuture(TransferDecision.releaseSource(released -> {
+                        assertEquals(context, released.context());
+                        order.add("release-second");
+                        return CompletableFuture.completedFuture(null);
+                    }));
+                });
+            }
+        };
+        try (var host = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(1),
+                Duration.ofSeconds(1))) {
+            host.load(List.of(first, second));
+            host.enable();
+            var selected = host.selectTransferPreparation();
+            assertTrue(selected != null);
+            PreparedTransfer prepared = selected.prepare(new TransferPreparingEvent(context))
+                    .toCompletableFuture().get(5, TimeUnit.SECONDS);
+            assertTrue(prepared.requiresSourceRelease());
+            assertEquals(List.of("prepare-first", "prepare-second"), order);
+            prepared.sourceReleased(new SourceReleasedEvent(context)).toCompletableFuture()
+                    .get(5, TimeUnit.SECONDS);
+            assertEquals(List.of("prepare-first", "prepare-second", "release-first", "release-second"), order);
+        }
+    }
+
+    @Test
+    void transferSelectionSkipsEmptyCohortsAndFailsWhenPinnedOwnerIsRevoked() throws Exception {
+        try (var empty = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(1))) {
+            empty.load(List.of());
+            empty.enable();
+            assertEquals(null, empty.selectTransferPreparation());
+        }
+
+        var subscription = new java.util.concurrent.atomic.AtomicReference<EventSubscription>();
+        var calls = new AtomicInteger();
+        var plugin = new Plugin() {
+            @Override public void onLoad(PluginContext pluginContext) {
+                subscription.set(pluginContext.events().subscribe(TransferPreparingEvent.class, event -> {
+                    calls.incrementAndGet();
+                    return CompletableFuture.completedFuture(TransferDecision.allow());
+                }));
+            }
+        };
+        try (var host = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(1))) {
+            host.load(List.of(plugin));
+            host.enable();
+            var selected = host.selectTransferPreparation();
+            subscription.get().close();
+            assertThrows(CompletionException.class, () -> selected.prepare(
+                    new TransferPreparingEvent(transferContext())).toCompletableFuture().join());
+            assertEquals(0, calls.get());
+        }
+    }
+
+    @Test
+    void transferReleaseCallbackCannotOutliveItsOwner() throws Exception {
+        var subscription = new java.util.concurrent.atomic.AtomicReference<EventSubscription>();
+        var invoked = new AtomicInteger();
+        var plugin = new Plugin() {
+            @Override public void onLoad(PluginContext pluginContext) {
+                subscription.set(pluginContext.events().subscribe(TransferPreparingEvent.class, event ->
+                        CompletableFuture.completedFuture(TransferDecision.releaseSource(released -> {
+                            invoked.incrementAndGet();
+                            return CompletableFuture.completedFuture(null);
+                        }))));
+            }
+        };
+        try (var host = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(1))) {
+            host.load(List.of(plugin));
+            host.enable();
+            TransferContext context = transferContext();
+            PreparedTransfer prepared = host.selectTransferPreparation()
+                    .prepare(new TransferPreparingEvent(context)).toCompletableFuture().get(5, TimeUnit.SECONDS);
+            subscription.get().close();
+            assertThrows(CompletionException.class, () -> prepared.sourceReleased(
+                    new SourceReleasedEvent(context)).toCompletableFuture().join());
+            assertEquals(0, invoked.get());
+        }
+    }
+
+    @Test
+    void sourceConfirmationUsesOneDeadlineForTheWholePinnedCohort() throws Exception {
+        var firstStarted = new CountDownLatch(1);
+        var laterCallback = new AtomicInteger();
+        var pending = new CompletableFuture<Void>();
+        var plugin = new Plugin() {
+            @Override public void onLoad(PluginContext pluginContext) {
+                pluginContext.events().subscribe(TransferPreparingEvent.class, event ->
+                        CompletableFuture.completedFuture(TransferDecision.releaseSource(released -> {
+                            firstStarted.countDown();
+                            return pending;
+                        })));
+                pluginContext.events().subscribe(TransferPreparingEvent.class, event ->
+                        CompletableFuture.completedFuture(TransferDecision.releaseSource(released -> {
+                            laterCallback.incrementAndGet();
+                            return CompletableFuture.completedFuture(null);
+                        })));
+            }
+        };
+        try (var host = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(1),
+                Duration.ofMillis(80))) {
+            host.load(List.of(plugin));
+            host.enable();
+            TransferContext context = transferContext();
+            PreparedTransfer prepared = host.selectTransferPreparation().prepare(new TransferPreparingEvent(context))
+                    .toCompletableFuture().get(5, TimeUnit.SECONDS);
+            assertThrows(CompletionException.class, () -> prepared.sourceReleased(new SourceReleasedEvent(context))
+                    .toCompletableFuture().join());
+            assertTrue(firstStarted.await(1, TimeUnit.SECONDS));
+            assertEquals(0, laterCallback.get());
+        }
+    }
+
+    @Test
+    void cancellationOfSourceConfirmationStopsLaterPinnedCallbacks() throws Exception {
+        var firstStarted = new CountDownLatch(1);
+        var laterCallback = new AtomicInteger();
+        var pending = new CompletableFuture<Void>();
+        var plugin = new Plugin() {
+            @Override public void onLoad(PluginContext pluginContext) {
+                pluginContext.events().subscribe(TransferPreparingEvent.class, event ->
+                        CompletableFuture.completedFuture(TransferDecision.releaseSource(released -> {
+                            firstStarted.countDown();
+                            return pending;
+                        })));
+                pluginContext.events().subscribe(TransferPreparingEvent.class, event ->
+                        CompletableFuture.completedFuture(TransferDecision.releaseSource(released -> {
+                            laterCallback.incrementAndGet();
+                            return CompletableFuture.completedFuture(null);
+                        })));
+            }
+        };
+        try (var host = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(1),
+                Duration.ofSeconds(5))) {
+            host.load(List.of(plugin));
+            host.enable();
+            TransferContext context = transferContext();
+            PreparedTransfer prepared = host.selectTransferPreparation().prepare(new TransferPreparingEvent(context))
+                    .toCompletableFuture().get(5, TimeUnit.SECONDS);
+            var confirmation = prepared.sourceReleased(new SourceReleasedEvent(context)).toCompletableFuture();
+            assertTrue(firstStarted.await(1, TimeUnit.SECONDS));
+            assertTrue(confirmation.cancel(false));
+            assertEquals(0, laterCallback.get());
+        }
+    }
+
+    @Test
+    void revokingEarlierRequiredParticipantWhileLaterCallbackWaitsFailsConfirmation() throws Exception {
+        var firstSubscription = new java.util.concurrent.atomic.AtomicReference<EventSubscription>();
+        var laterStarted = new CountDownLatch(1);
+        var laterPending = new CompletableFuture<Void>();
+        var first = new Plugin() {
+            @Override public void onLoad(PluginContext pluginContext) {
+                firstSubscription.set(pluginContext.events().subscribe(TransferPreparingEvent.class, event ->
+                        CompletableFuture.completedFuture(TransferDecision.releaseSource(released ->
+                                CompletableFuture.completedFuture(null)))));
+            }
+        };
+        var second = new Plugin() {
+            @Override public void onLoad(PluginContext pluginContext) {
+                pluginContext.events().subscribe(TransferPreparingEvent.class, event ->
+                        CompletableFuture.completedFuture(TransferDecision.releaseSource(released -> {
+                            laterStarted.countDown();
+                            return laterPending;
+                        })));
+            }
+        };
+        try (var host = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(1),
+                Duration.ofSeconds(2))) {
+            host.load(List.of(first, second));
+            host.enable();
+            TransferContext context = transferContext();
+            PreparedTransfer prepared = host.selectTransferPreparation().prepare(new TransferPreparingEvent(context))
+                    .toCompletableFuture().get(5, TimeUnit.SECONDS);
+            var confirmation = prepared.sourceReleased(new SourceReleasedEvent(context)).toCompletableFuture();
+            assertTrue(laterStarted.await(1, TimeUnit.SECONDS));
+            firstSubscription.get().close();
+            laterPending.complete(null);
+            assertThrows(CompletionException.class, confirmation::join);
+        }
+    }
+
+    private static TransferContext transferContext() {
+        return new TransferContext(UUID.randomUUID(), PLAYER.identity(),
+                new dev.moonbridge.api.ServerView("source", URI.create("tcp://127.0.0.1:25565"), Map.of(), Map.of()),
+                new dev.moonbridge.api.ServerView("target", URI.create("tcp://127.0.0.1:25566"), Map.of(), Map.of()),
+                1, 2, 3, 4, Instant.now().plusSeconds(10));
+    }
+
     @Test
     void accessChecksFailClosedOnTimeoutAndSkipWorkersWhenNoPluginListens() throws Exception {
         var hostWithoutChecks = new PluginHost(new InMemoryBackendCatalog(), players(), Duration.ofSeconds(1),

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

@@ -5,8 +5,10 @@ import dev.moonbridge.api.PlacementDecision;
 import dev.moonbridge.api.TransferStatus;
 import dev.moonbridge.api.event.Event;
 import dev.moonbridge.api.event.PlayerDisconnectedEvent;
+import dev.moonbridge.api.event.SourceReleasedEvent;
 import dev.moonbridge.api.event.ServerConnectedEvent;
 import dev.moonbridge.core.event.EventDispatcher;
+import dev.moonbridge.core.event.PreparedTransfer;
 import dev.moonbridge.core.backend.BackendId;
 import dev.moonbridge.core.backend.BackendCatalog;
 import dev.moonbridge.core.backend.BackendHandle;
@@ -165,6 +167,7 @@ final class SessionTransferTest {
         var catalog = new InMemoryBackendCatalog();
         try (ServerSocket oldServer = server(); ServerSocket newServer = server()) {
             var oldReceived = new CompletableFuture<Void>();
+            var oldClosed = new CompletableFuture<Void>();
             var newClosed = new CompletableFuture<Void>();
             backendThread(oldServer, () -> {
                 try (Socket socket = oldServer.accept()) {
@@ -176,24 +179,21 @@ final class SessionTransferTest {
                     assertArrayEquals(new byte[]{0x01, 0x66}, readFrame(input));
                     oldReceived.complete(null);
                     while (input.read() != -1) { }
-                } catch (Throwable failure) { oldReceived.completeExceptionally(failure); }
+                    oldClosed.complete(null);
+                } catch (Throwable failure) { oldReceived.completeExceptionally(failure); oldClosed.completeExceptionally(failure); }
             });
             backendThread(newServer, () -> {
                 try (Socket socket = newServer.accept()) {
                     socket.setSoTimeout(5000);
-                    var input = new DataInputStream(socket.getInputStream());
-                    var output = new DataOutputStream(socket.getOutputStream());
-                    acceptHandshakeAndLogin(input, output);
-                    sendLoginSuccess(output);
-                    sendJoinGame(output, 0, 200);
-                    writeFrame(output, new byte[]{0x08});
-                    assertEquals(-1, input.read());
+                    assertEquals(-1, socket.getInputStream().read(),
+                            "a rejected pre-detach candidate must not receive even a Login Start");
                     newClosed.complete(null);
                 } catch (Throwable failure) { newClosed.completeExceptionally(failure); }
             });
             register(catalog, "old", oldServer);
             register(catalog, "new", newServer);
             var listener = listener(catalog);
+            listener.setEvents(releaseSourceEvents(), Duration.ofSeconds(5));
             try {
                 int port = ((InetSocketAddress) listener.start().toCompletableFuture()
                         .get(5, TimeUnit.SECONDS).localAddress()).getPort();
@@ -273,6 +273,8 @@ final class SessionTransferTest {
                             Unpooled.wrappedBuffer(new byte[]{0x66}))).get(5, TimeUnit.SECONDS);
                     oldReceived.get(5, TimeUnit.SECONDS);
                     newClosed.get(5, TimeUnit.SECONDS);
+                    assertThrows(TimeoutException.class, () -> oldClosed.get(150, TimeUnit.MILLISECONDS),
+                            "partial-frame rejection occurs before source close");
                 }
             } finally { listener.close().toCompletableFuture().get(5, TimeUnit.SECONDS); }
         }
@@ -1019,6 +1021,7 @@ final class SessionTransferTest {
             register(catalog, "old", oldServer);
             register(catalog, "new", newServer);
             var listener = listener(catalog);
+            listener.setEvents(releaseSourceEvents(), Duration.ofSeconds(5));
             try {
                 var bound = (InetSocketAddress) listener.start().toCompletableFuture()
                         .get(5, TimeUnit.SECONDS).localAddress();
@@ -1505,6 +1508,23 @@ final class SessionTransferTest {
         return listener;
     }
 
+    private static EventDispatcher releaseSourceEvents() {
+        return new EventDispatcher() {
+            @Override public boolean hasSubscribers(Class<?> eventType) { return false; }
+            @Override public <R> CompletionStage<R> dispatch(Event<R> event) {
+                return CompletableFuture.completedFuture(null);
+            }
+            @Override public dev.moonbridge.core.event.TransferPreparation selectTransferPreparation() {
+                return ignored -> CompletableFuture.completedFuture(new PreparedTransfer() {
+                    @Override public boolean requiresSourceRelease() { return true; }
+                    @Override public CompletionStage<Void> sourceReleased(SourceReleasedEvent event) {
+                        return CompletableFuture.completedFuture(null);
+                    }
+                });
+            }
+        };
+    }
+
     private static dev.moonbridge.api.PlayerView awaitPlayer(ProxySessionListener listener) throws Exception {
         long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
         while (listener.online().isEmpty() && System.nanoTime() < deadline) Thread.sleep(5);

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

@@ -0,0 +1,522 @@
+package dev.moonbridge.core.session;
+
+import dev.moonbridge.api.PlacementDecision;
+import dev.moonbridge.api.PlayerView;
+import dev.moonbridge.api.TransferStatus;
+import dev.moonbridge.api.event.Event;
+import dev.moonbridge.api.event.SourceReleasedEvent;
+import dev.moonbridge.api.event.TransferDecision;
+import dev.moonbridge.api.event.TransferPreparingEvent;
+import dev.moonbridge.core.backend.BackendId;
+import dev.moonbridge.core.backend.BackendOwner;
+import dev.moonbridge.core.backend.BackendRegistration;
+import dev.moonbridge.core.backend.InMemoryBackendCatalog;
+import dev.moonbridge.core.event.EventDispatcher;
+import dev.moonbridge.core.event.PreparedTransfer;
+import dev.moonbridge.core.event.TransferPreparation;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.EOFException;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+final class SourceReleaseTransferTest {
+    private static final String USERNAME = "ReleasePlayer";
+    private static final byte[] BUFFERED_CLIENT_FRAME = {0x03, 0x55};
+
+    @Test
+    void waitsForSaveAcknowledgementAfterExactSourceEofBeforeTargetLoginAndDiscardsFrames() throws Exception {
+        var acknowledgement = new CompletableFuture<Void>();
+        var callbackEntered = new CompletableFuture<Void>();
+        var releasedEvent = new AtomicReference<SourceReleasedEvent>();
+        try (TransferFixture fixture = new TransferFixture(preparation(releaseSource(event -> {
+            releasedEvent.set(event);
+            callbackEntered.complete(null);
+            return acknowledgement;
+        })))) {
+            try (Client client = fixture.login()) {
+                PlayerView player = fixture.player();
+                var transfer = fixture.listener.transfer(player.identity(), "target").toCompletableFuture();
+                fixture.targetAccepted.get(5, TimeUnit.SECONDS);
+                fixture.sourceClosed.get(5, TimeUnit.SECONDS);
+                callbackEntered.get(5, TimeUnit.SECONDS);
+                assertNotNull(releasedEvent.get(), "source release callback must follow the source socket close");
+                assertEquals("source", releasedEvent.get().context().source().name());
+                PlayerView releasedView = fixture.listener.find(player.identity()).orElseThrow();
+                assertTrue(releasedView.currentServer().isEmpty(), "source EOF must clear the current backend view");
+                assertTrue(fixture.listener.online().stream().anyMatch(online -> online.identity().equals(player.identity())),
+                        "source EOF must not emit a player-disconnected state while the frontend remains connected");
+                assertFalse(fixture.targetLogin.isDone(), "target Login Start must wait for the save acknowledgement");
+                assertFalse(transfer.isDone());
+
+                writeFrame(client.output(), BUFFERED_CLIENT_FRAME);
+                client.setSoTimeout(150);
+                assertThrows(SocketTimeoutException.class, () -> client.input().read(),
+                        "source EOF must not be forwarded to the client while the release callback waits");
+                assertFalse(fixture.targetLogin.isDone());
+
+                acknowledgement.complete(null);
+                fixture.targetLogin.get(5, TimeUnit.SECONDS);
+                assertEquals(TransferStatus.NETWORK_READY, transfer.get(5, TimeUnit.SECONDS).status());
+                client.setSoTimeout(5000);
+                assertEquals(7, packetId(readFrame(client.input())));
+                assertEquals(7, packetId(readFrame(client.input())));
+                assertEquals(8, packetId(readFrame(client.input())),
+                        "the candidate's Position and Look must reach the client after cutover");
+                assertThrows(java.util.concurrent.TimeoutException.class,
+                        () -> fixture.targetFirstClientFrame.get(200, TimeUnit.MILLISECONDS),
+                        "the frame read while the source-release gate is held must not be replayed");
+            }
+        }
+    }
+
+    @Test
+    void preparationDenialBeforeReleaseKeepsTheSourceRelayUsable() throws Exception {
+        var rejected = new CompletableFuture<PreparedTransfer>();
+        rejected.completeExceptionally(new IllegalStateException("transfer denied"));
+        try (TransferFixture fixture = new TransferFixture(event -> rejected)) {
+            try (Client client = fixture.login()) {
+                PlayerView player = fixture.player();
+                var result = fixture.listener.transfer(player.identity(), "target").toCompletableFuture()
+                        .get(5, TimeUnit.SECONDS);
+                assertEquals(TransferStatus.FAILED, result.status());
+                assertEquals("source", fixture.listener.find(player.identity())
+                        .flatMap(PlayerView::currentServer).orElseThrow());
+                assertThrows(java.util.concurrent.TimeoutException.class,
+                        () -> fixture.sourceClosed.get(150, TimeUnit.MILLISECONDS));
+                writeFrame(fixture.sourceOutput.get(), new byte[]{0x03, 0x44});
+                assertArrayEquals(new byte[]{0x03, 0x44}, readFrame(client.input()));
+                assertFalse(fixture.targetLogin.isDone(), "denial must keep target Login Start deferred");
+            }
+        }
+    }
+
+    @Test
+    void sourceReleaseCallbackFailureDisconnectsAfterSourceEof() throws Exception {
+        try (TransferFixture fixture = new TransferFixture(preparation(releaseSource(
+                event -> CompletableFuture.failedFuture(new IllegalStateException("save failed")))))) {
+            try (Client client = fixture.login()) {
+                PlayerView player = fixture.player();
+                var transfer = fixture.listener.transfer(player.identity(), "target").toCompletableFuture();
+                fixture.sourceClosed.get(5, TimeUnit.SECONDS);
+                assertEquals(TransferStatus.FAILED, transfer.get(5, TimeUnit.SECONDS).status());
+                assertEquals(-1, client.input().read());
+                assertFalse(fixture.targetLogin.isDone(), "failed source save must prevent target Login Start");
+            }
+        }
+    }
+
+    @Test
+    void disconnectMakesLatePreparationCompletionStale() throws Exception {
+        var pending = new CompletableFuture<PreparedTransfer>();
+        var sourceReleasedCalls = new AtomicInteger();
+        try (TransferFixture fixture = new TransferFixture(event -> pending)) {
+            Client client = fixture.login();
+            try {
+                PlayerView player = fixture.player();
+                fixture.listener.transfer(player.identity(), "target");
+                fixture.targetAccepted.get(5, TimeUnit.SECONDS);
+                client.close();
+                fixture.sourceClosed.get(5, TimeUnit.SECONDS);
+                pending.complete(releaseSource(event -> {
+                    sourceReleasedCalls.incrementAndGet();
+                    return CompletableFuture.completedFuture(null);
+                }));
+                Thread.sleep(100);
+                assertEquals(0, sourceReleasedCalls.get(), "a stale callback must not reopen the transfer");
+                assertFalse(fixture.targetLogin.isDone(), "a disconnected session must not start destination login");
+            } finally {
+                client.close();
+            }
+        }
+    }
+
+    @Test
+    void targetRegistrationReplacementWhileSaveAcknowledgementWaitsDisconnectsWithoutLoginStart() throws Exception {
+        var acknowledgement = new CompletableFuture<Void>();
+        var callbackEntered = new CompletableFuture<Void>();
+        try (TransferFixture fixture = new TransferFixture(preparation(releaseSource(
+                event -> { callbackEntered.complete(null); return acknowledgement; })))) {
+            try (Client client = fixture.login()) {
+                PlayerView player = fixture.player();
+                var transfer = fixture.listener.transfer(player.identity(), "target").toCompletableFuture();
+                fixture.targetAccepted.get(5, TimeUnit.SECONDS);
+                fixture.sourceClosed.get(5, TimeUnit.SECONDS);
+                callbackEntered.get(5, TimeUnit.SECONDS);
+                fixture.replaceTargetRegistration();
+                acknowledgement.complete(null);
+                assertEquals(TransferStatus.FAILED, transfer.get(5, TimeUnit.SECONDS).status());
+                assertEquals(-1, client.input().read());
+                assertFalse(fixture.targetLogin.isDone(), "stale target generation must not receive Login Start");
+            }
+        }
+    }
+
+    @Test
+    void frontendDisconnectDuringReleaseConfirmationCancelsItAndNeverStartsTargetLogin() throws Exception {
+        var acknowledgement = new CompletableFuture<Void>();
+        var callbackEntered = new CompletableFuture<Void>();
+        var cancelled = new CompletableFuture<Void>();
+        acknowledgement.whenComplete((ignored, failure) -> {
+            if (acknowledgement.isCancelled()) cancelled.complete(null);
+        });
+        try (TransferFixture fixture = new TransferFixture(preparation(releaseSource(
+                event -> { callbackEntered.complete(null); return acknowledgement; })))) {
+            Client client = fixture.login();
+            try {
+                PlayerView player = fixture.player();
+                fixture.listener.transfer(player.identity(), "target");
+                fixture.targetAccepted.get(5, TimeUnit.SECONDS);
+                fixture.sourceClosed.get(5, TimeUnit.SECONDS);
+                callbackEntered.get(5, TimeUnit.SECONDS);
+                client.close();
+                cancelled.get(5, TimeUnit.SECONDS);
+                acknowledgement.complete(null);
+                assertFalse(fixture.targetLogin.isDone(), "a stale release callback must not start target login");
+            } finally {
+                client.close();
+            }
+        }
+    }
+
+    @Test
+    void candidateDeathBeforeSourceReleaseLeavesSourceUsable() throws Exception {
+        var pending = new CompletableFuture<PreparedTransfer>();
+        try (TransferFixture fixture = new TransferFixture(event -> pending)) {
+            try (Client client = fixture.login()) {
+                PlayerView player = fixture.player();
+                var transfer = fixture.listener.transfer(player.identity(), "target").toCompletableFuture();
+                fixture.targetAccepted.get(5, TimeUnit.SECONDS);
+                writeFrame(client.output(), BUFFERED_CLIENT_FRAME);
+                assertArrayEquals(BUFFERED_CLIENT_FRAME, fixture.sourceClientFrames.poll(5, TimeUnit.SECONDS),
+                        "the source relay remains active while preparation is pending");
+                fixture.disconnectTarget();
+                assertEquals(TransferStatus.FAILED, transfer.get(5, TimeUnit.SECONDS).status());
+                assertThrows(java.util.concurrent.TimeoutException.class,
+                        () -> fixture.sourceClosed.get(150, TimeUnit.MILLISECONDS));
+                assertEquals("source", fixture.listener.find(player.identity()).flatMap(PlayerView::currentServer)
+                        .orElseThrow());
+                writeFrame(fixture.sourceOutput.get(), new byte[]{0x03, 0x66});
+                assertArrayEquals(new byte[]{0x03, 0x66}, readFrame(client.input()));
+                assertFalse(fixture.targetLogin.isDone());
+            }
+        }
+    }
+
+    @Test
+    void candidateDeathAfterSourceReleaseDisconnectsTheClient() throws Exception {
+        var acknowledgement = new CompletableFuture<Void>();
+        var callbackEntered = new CompletableFuture<Void>();
+        var cancelled = new CompletableFuture<Void>();
+        acknowledgement.whenComplete((ignored, failure) -> {
+            if (acknowledgement.isCancelled()) cancelled.complete(null);
+        });
+        try (TransferFixture fixture = new TransferFixture(preparation(releaseSource(
+                event -> { callbackEntered.complete(null); return acknowledgement; })))) {
+            try (Client client = fixture.login()) {
+                PlayerView player = fixture.player();
+                var transfer = fixture.listener.transfer(player.identity(), "target").toCompletableFuture();
+                fixture.targetAccepted.get(5, TimeUnit.SECONDS);
+                fixture.sourceClosed.get(5, TimeUnit.SECONDS);
+                callbackEntered.get(5, TimeUnit.SECONDS);
+                fixture.disconnectTarget();
+                assertEquals(TransferStatus.FAILED, transfer.get(5, TimeUnit.SECONDS).status());
+                assertEquals(-1, client.input().read());
+                cancelled.get(5, TimeUnit.SECONDS);
+                assertTrue(acknowledgement.isCancelled(), "candidate loss must cancel a pending release callback");
+                assertFalse(fixture.targetLogin.isDone());
+            }
+        }
+    }
+
+    private static PreparedTransfer releaseSource(SourceReleasedHandler handler) {
+        return new PreparedTransfer() {
+            @Override public boolean requiresSourceRelease() { return true; }
+            @Override public CompletionStage<Void> sourceReleased(SourceReleasedEvent event) {
+                return handler.onSourceReleased(event);
+            }
+        };
+    }
+
+    @FunctionalInterface
+    private interface SourceReleasedHandler {
+        CompletionStage<Void> onSourceReleased(SourceReleasedEvent event);
+    }
+
+    private static TransferPreparation preparation(PreparedTransfer prepared) {
+        return event -> CompletableFuture.completedFuture(prepared);
+    }
+
+    private static EventDispatcher dispatcher(TransferPreparation preparation) {
+        return new EventDispatcher() {
+            @Override public boolean hasSubscribers(Class<?> eventType) { return false; }
+            @Override public <R> CompletionStage<R> dispatch(Event<R> event) {
+                return CompletableFuture.completedFuture(null);
+            }
+            @Override public TransferPreparation selectTransferPreparation() { return preparation; }
+        };
+    }
+
+    private static final class TransferFixture implements AutoCloseable {
+        private final ServerSocket source = server();
+        private final ServerSocket target = server();
+        private final ServerSocket replacementTarget = server();
+        private final CompletableFuture<Void> sourceClosed = new CompletableFuture<>();
+        private final CompletableFuture<Void> targetAccepted = new CompletableFuture<>();
+        private final CompletableFuture<String> targetLogin = new CompletableFuture<>();
+        private final CompletableFuture<byte[]> targetFirstClientFrame = new CompletableFuture<>();
+        private final LinkedBlockingQueue<byte[]> sourceClientFrames = new LinkedBlockingQueue<>();
+        private final AtomicReference<DataOutputStream> sourceOutput = new AtomicReference<>();
+        private final AtomicReference<Socket> targetSocket = new AtomicReference<>();
+        private final ProxySessionListener listener;
+        private final int proxyPort;
+        private final InMemoryBackendCatalog catalog = new InMemoryBackendCatalog();
+
+        private TransferFixture(TransferPreparation preparation) throws Exception {
+            backendThread(source, this::serveSource);
+            backendThread(target, this::serveTarget);
+            register(catalog, "source", source);
+            register(catalog, "target", target);
+            listener = new ProxySessionListener(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), catalog);
+            listener.setPlacement(ignored -> CompletableFuture.completedFuture(Optional.of(
+                    PlacementDecision.select("source"))));
+            listener.setEvents(dispatcher(preparation), Duration.ofSeconds(5));
+            proxyPort = ((InetSocketAddress) listener.start().toCompletableFuture()
+                    .get(5, TimeUnit.SECONDS).localAddress()).getPort();
+        }
+
+        private Client login() throws Exception {
+            Client client = new Client(proxyPort);
+            client.login(USERNAME);
+            assertEquals(2, packetId(readFrame(client.input())));
+            assertEquals(1, packetId(readFrame(client.input())));
+            assertEquals(8, packetId(readFrame(client.input())));
+            player();
+            return client;
+        }
+
+        private PlayerView player() throws Exception {
+            long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+            while (System.nanoTime() < deadline) {
+                for (PlayerView player : listener.online()) if (player.username().equals(USERNAME)) return player;
+                Thread.sleep(2);
+            }
+            throw new IOException("player did not complete login");
+        }
+
+        private void replaceTargetRegistration() {
+            catalog.register(new BackendRegistration(new BackendId("target"), new BackendOwner("static", 0),
+                    URI.create("tcp://127.0.0.1:" + replacementTarget.getLocalPort()), Map.of(), Map.of()));
+        }
+
+        private void disconnectTarget() throws Exception {
+            targetAccepted.get(5, TimeUnit.SECONDS);
+            Socket socket = targetSocket.get();
+            if (socket == null) throw new IOException("target socket was not published");
+            socket.close();
+        }
+
+        private void serveSource() {
+            try (Socket socket = source.accept()) {
+                socket.setSoTimeout(5000);
+                DataInputStream input = new DataInputStream(socket.getInputStream());
+                DataOutputStream output = new DataOutputStream(socket.getOutputStream());
+                acceptLogin(input, output, USERNAME, 42);
+                sourceOutput.set(output);
+                while (true) {
+                    try { sourceClientFrames.add(readFrame(input)); }
+                    catch (EOFException closed) { break; }
+                }
+                sourceClosed.complete(null);
+            } catch (Throwable failure) { sourceClosed.completeExceptionally(failure); }
+        }
+
+        private void serveTarget() {
+            try (Socket socket = target.accept()) {
+                targetSocket.set(socket);
+                socket.setSoTimeout(5000);
+                targetAccepted.complete(null);
+                DataInputStream input = new DataInputStream(socket.getInputStream());
+                DataOutputStream output = new DataOutputStream(socket.getOutputStream());
+                readFrame(input); // Backend handshake.
+                byte[] login = readFrame(input);
+                ByteArrayInputStream loginData = new ByteArrayInputStream(login);
+                assertEquals(0, readVarInt(loginData));
+                String username = readString(loginData);
+                targetLogin.complete(username);
+                writeFrameBundle(output, loginSuccess(username), joinGame(99), new byte[]{0x08});
+                byte[] frame = readFrame(input);
+                targetFirstClientFrame.complete(frame);
+                while (input.read() != -1) { }
+            } catch (Throwable failure) {
+                targetAccepted.completeExceptionally(failure);
+            }
+        }
+
+        @Override public void close() throws Exception {
+            try { listener.close().toCompletableFuture().get(5, TimeUnit.SECONDS); }
+            finally {
+                source.close();
+                target.close();
+                replacementTarget.close();
+            }
+        }
+    }
+
+    private static final class Client implements AutoCloseable {
+        private final Socket socket = new Socket();
+        private final DataInputStream input;
+        private final DataOutputStream output;
+        private Client(int port) throws IOException {
+            socket.setSoTimeout(5000);
+            socket.connect(new InetSocketAddress(InetAddress.getLoopbackAddress(), port), 5000);
+            input = new DataInputStream(socket.getInputStream());
+            output = new DataOutputStream(socket.getOutputStream());
+        }
+        private void login(String username) throws IOException { sendHandshakeAndLogin(output, username); }
+        private DataInputStream input() { return input; }
+        private DataOutputStream output() { return output; }
+        private void setSoTimeout(int timeoutMillis) throws IOException { socket.setSoTimeout(timeoutMillis); }
+        @Override public void close() throws IOException { socket.close(); }
+    }
+
+    private static ServerSocket server() throws IOException {
+        return new ServerSocket(0, 8, InetAddress.getLoopbackAddress());
+    }
+
+    private static void backendThread(ServerSocket socket, Runnable task) {
+        Thread thread = new Thread(task, "source-release-test-backend-" + socket.getLocalPort());
+        thread.setDaemon(true);
+        thread.start();
+    }
+
+    private static void register(InMemoryBackendCatalog catalog, String name, ServerSocket socket) {
+        catalog.register(new BackendRegistration(new BackendId(name), new BackendOwner("static", 0),
+                URI.create("tcp://127.0.0.1:" + socket.getLocalPort()), Map.of(), Map.of()));
+    }
+
+    private static void acceptLogin(DataInputStream input, DataOutputStream output, String username, int entityId)
+            throws Exception {
+        readFrame(input); // Handshake.
+        byte[] login = readFrame(input);
+        ByteArrayInputStream loginData = new ByteArrayInputStream(login);
+        assertEquals(0, readVarInt(loginData));
+        assertEquals(username, readString(loginData));
+        sendLoginSuccess(output, username);
+        sendJoinGame(output, entityId);
+        writeFrame(output, new byte[]{0x08});
+    }
+
+    private static void sendLoginSuccess(DataOutputStream output, String username) throws IOException {
+        writeFrame(output, loginSuccess(username));
+    }
+
+    private static byte[] loginSuccess(String username) throws IOException {
+        ByteArrayOutputStream body = new ByteArrayOutputStream();
+        writeVarInt(body, 2);
+        writeString(body, UUID.nameUUIDFromBytes(("OfflinePlayer:" + username).getBytes(StandardCharsets.UTF_8)).toString());
+        writeString(body, username);
+        return body.toByteArray();
+    }
+
+    private static void sendJoinGame(DataOutputStream output, int entityId) throws IOException {
+        writeFrame(output, joinGame(entityId));
+    }
+
+    private static byte[] joinGame(int entityId) throws IOException {
+        ByteArrayOutputStream body = new ByteArrayOutputStream();
+        writeVarInt(body, 1);
+        DataOutputStream data = new DataOutputStream(body);
+        data.writeInt(entityId); data.writeByte(0); data.writeByte(0); data.writeByte(0); data.writeByte(20);
+        writeString(body, "default");
+        return body.toByteArray();
+    }
+
+    private static void sendHandshakeAndLogin(DataOutputStream output, String username) throws IOException {
+        ByteArrayOutputStream handshake = new ByteArrayOutputStream();
+        writeVarInt(handshake, 0); writeVarInt(handshake, 5); writeString(handshake, "localhost");
+        handshake.write(0); handshake.write(255); writeVarInt(handshake, 2);
+        writeFrame(output, handshake.toByteArray());
+        ByteArrayOutputStream login = new ByteArrayOutputStream();
+        writeVarInt(login, 0); writeString(login, username);
+        writeFrame(output, login.toByteArray());
+    }
+
+    private static int packetId(byte[] frame) throws IOException { return readVarInt(new ByteArrayInputStream(frame)); }
+
+    private static byte[] readFrame(DataInputStream input) throws IOException {
+        int length = readVarInt(input);
+        if (length < 1 || length > 1_048_576) throw new IOException("invalid frame length " + length);
+        byte[] body = new byte[length]; input.readFully(body); return body;
+    }
+
+    private static void writeFrame(DataOutputStream output, byte[] body) throws IOException {
+        writeVarInt(output, body.length); output.write(body); output.flush();
+    }
+
+    private static void writeFrameBundle(DataOutputStream output, byte[]... bodies) throws IOException {
+        ByteArrayOutputStream bundle = new ByteArrayOutputStream();
+        for (byte[] body : bodies) { writeVarInt(bundle, body.length); bundle.write(body); }
+        output.write(bundle.toByteArray());
+        output.flush();
+    }
+
+    private static int readVarInt(java.io.InputStream input) throws IOException {
+        int result = 0;
+        for (int shift = 0; shift < 35; shift += 7) {
+            int next = input.read();
+            if (next < 0) throw new EOFException();
+            result |= (next & 0x7f) << shift;
+            if ((next & 0x80) == 0) return result;
+        }
+        throw new IOException("overlong VarInt");
+    }
+
+    private static void writeVarInt(java.io.OutputStream output, int value) throws IOException {
+        do {
+            int part = value & 0x7f; value >>>= 7;
+            output.write(value == 0 ? part : part | 0x80);
+        } while (value != 0);
+    }
+
+    private static String readString(ByteArrayInputStream input) throws IOException {
+        int length = readVarInt(input);
+        if (length < 0 || length > 1024) throw new IOException("invalid string length");
+        byte[] bytes = input.readNBytes(length);
+        if (bytes.length != length) throw new EOFException();
+        return new String(bytes, StandardCharsets.UTF_8);
+    }
+
+    private static void writeString(ByteArrayOutputStream output, String value) throws IOException {
+        byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+        writeVarInt(output, bytes.length); output.write(bytes);
+    }
+}

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

@@ -47,6 +47,30 @@ final class TransferFrameBufferTest {
         channel.finishAndReleaseAll();
     }
 
+    @Test
+    void discardModeReleasesQueuedFramesAndDropsFutureFramesWithoutReplay() {
+        var buffer = new TransferFrameBuffer(() -> { });
+        var channel = new EmbeddedChannel(buffer);
+        try {
+            ByteBuf queued = Unpooled.wrappedBuffer(new byte[]{1});
+            channel.writeInbound(queued);
+            channel.eventLoop().execute(buffer::discardFrames);
+            channel.runPendingTasks();
+            assertEquals(0, queued.refCnt());
+
+            ByteBuf later = Unpooled.wrappedBuffer(new byte[]{2});
+            channel.writeInbound(later);
+            assertEquals(0, later.refCnt());
+            assertNull(channel.readInbound());
+
+            channel.eventLoop().execute(buffer::drainAndRemove);
+            channel.runPendingTasks();
+            assertNull(channel.readInbound(), "removing the handler must not replay discarded source frames");
+        } finally {
+            channel.finishAndReleaseAll();
+        }
+    }
+
     @Test
     void exceedingCutoverLimitRejectsTheNewFrame() {
         var overflows = new AtomicInteger();

proxy-plugin-api/src/main/java/dev/moonbridge/api/event/Event.java

@@ -9,5 +9,5 @@ package dev.moonbridge.api.event;
  */
 public sealed interface Event<R>
         permits ConnectionAdmissionEvent, PlayerAdmissionEvent, ServerConnectedEvent,
-                PlayerDisconnectedEvent {
+                PlayerDisconnectedEvent, TransferPreparingEvent {
 }

proxy-plugin-api/src/main/java/dev/moonbridge/api/event/Events.java

@@ -12,7 +12,9 @@ package dev.moonbridge.api.event;
  * subject to the same configured deadline, including time waiting in the dispatch queue.
  * Each concrete event defines its result aggregation.
  * Admission events aggregate {@code AccessDecision} results by allowing access when every listener
- * allows it; the first denial denies access. With no listeners, admission is allowed. A listener
+ * allows it; the first denial denies access. Transfer preparation events similarly aggregate
+ * {@link TransferDecision} results; a denial is terminal, and release-source callbacks are retained
+ * in registration order. With no listeners, admission and transfer preparation are allowed. A listener
  * returning a null stage or result, failing, or timing out fails closed. Notification events use
  * {@link Void}; their listener results are ignored.
  *

proxy-plugin-api/src/main/java/dev/moonbridge/api/event/SourceReleasedEvent.java

@@ -0,0 +1,14 @@
+package dev.moonbridge.api.event;
+
+import java.util.Objects;
+
+/**
+ * Callback context indicating that the proxy closed the exact source backend socket. This does
+ * not assert that the backend fired a game Quit event or persisted any state; the handler confirms
+ * its own durable boundary by completing successfully.
+ */
+public record SourceReleasedEvent(TransferContext context) {
+    public SourceReleasedEvent {
+        Objects.requireNonNull(context, "context");
+    }
+}

proxy-plugin-api/src/main/java/dev/moonbridge/api/event/SourceReleasedHandler.java

@@ -0,0 +1,10 @@
+package dev.moonbridge.api.event;
+
+import java.util.concurrent.CompletionStage;
+
+/** Confirms a participant's durable boundary after the source backend socket has closed. */
+@FunctionalInterface
+public interface SourceReleasedHandler {
+    /** Successful completion confirms the participant's durable boundary. */
+    CompletionStage<Void> onSourceReleased(SourceReleasedEvent event);
+}

proxy-plugin-api/src/main/java/dev/moonbridge/api/event/TransferContext.java

@@ -0,0 +1,33 @@
+package dev.moonbridge.api.event;
+
+import dev.moonbridge.api.PlayerIdentity;
+import dev.moonbridge.api.ServerView;
+import java.time.Instant;
+import java.util.Objects;
+import java.util.UUID;
+
+/** Immutable identity and registration snapshot for one online backend transfer. */
+public record TransferContext(
+        UUID transferId,
+        PlayerIdentity player,
+        ServerView source,
+        ServerView target,
+        long sourceRegistrationGeneration,
+        long targetRegistrationGeneration,
+        long sourceEpoch,
+        long targetEpoch,
+        Instant deadline) {
+    public TransferContext {
+        Objects.requireNonNull(transferId, "transferId");
+        Objects.requireNonNull(player, "player");
+        Objects.requireNonNull(source, "source");
+        Objects.requireNonNull(target, "target");
+        Objects.requireNonNull(deadline, "deadline");
+        if (sourceRegistrationGeneration < 1 || targetRegistrationGeneration < 1) {
+            throw new IllegalArgumentException("registration generations must be positive");
+        }
+        if (sourceEpoch < 0 || targetEpoch < 0) {
+            throw new IllegalArgumentException("backend epochs must be non-negative");
+        }
+    }
+}

proxy-plugin-api/src/main/java/dev/moonbridge/api/event/TransferDecision.java

@@ -0,0 +1,38 @@
+package dev.moonbridge.api.event;
+
+import java.util.Objects;
+
+/** One listener's response to a transfer preparation request. Denial detail is not sent to clients. */
+public sealed interface TransferDecision permits TransferDecision.Allowed,
+        TransferDecision.Denied, TransferDecision.ReleaseSource {
+    static TransferDecision allow() {
+        return Allowed.instance();
+    }
+
+    static TransferDecision deny(String reason) {
+        return new Denied(reason);
+    }
+
+    static TransferDecision releaseSource(SourceReleasedHandler handler) {
+        return new ReleaseSource(handler);
+    }
+
+    record Allowed() implements TransferDecision {
+        private static final Allowed ALLOWED = new Allowed();
+
+        private static Allowed instance() { return ALLOWED; }
+    }
+
+    record Denied(String reason) implements TransferDecision {
+        public Denied {
+            Objects.requireNonNull(reason, "reason");
+            if (reason.isBlank()) throw new IllegalArgumentException("reason must not be blank");
+        }
+    }
+
+    record ReleaseSource(SourceReleasedHandler handler) implements TransferDecision {
+        public ReleaseSource {
+            Objects.requireNonNull(handler, "handler");
+        }
+    }
+}

proxy-plugin-api/src/main/java/dev/moonbridge/api/event/TransferPreparingEvent.java

@@ -0,0 +1,16 @@
+package dev.moonbridge.api.event;
+
+import java.util.Objects;
+
+/**
+ * Asynchronous decision phase after the proxy has connected to the target backend by TCP and before
+ * it writes the backend handshake or login frames. Listeners may allow, deny, or register a durable
+ * source-release callback. Preparation must not mutate state or capture/reset the player in a way
+ * that requires rollback: the source remains active during this phase and transfer failure leaves
+ * it active.
+ */
+public record TransferPreparingEvent(TransferContext context) implements Event<TransferDecision> {
+    public TransferPreparingEvent {
+        Objects.requireNonNull(context, "context");
+    }
+}

proxy-plugin-api/src/test/java/dev/moonbridge/api/event/EventsApiTest.java

@@ -6,8 +6,11 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
 import dev.moonbridge.api.AccessDecision;
 import dev.moonbridge.api.PlayerIdentity;
 import dev.moonbridge.api.PlayerView;
+import dev.moonbridge.api.ServerView;
 import java.net.InetSocketAddress;
 import java.util.Optional;
+import java.time.Instant;
+import java.net.URI;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import org.junit.jupiter.api.Test;
@@ -53,4 +56,23 @@ class EventsApiTest {
         notification.close();
     }
 
+    @Test
+    void transferApiCarriesExactImmutableContextAndDecisionCallbacks() {
+        var source = new ServerView("source", URI.create("tcp://127.0.0.1:25565"), java.util.Map.of(), java.util.Map.of());
+        var target = new ServerView("target", URI.create("tcp://127.0.0.1:25566"), java.util.Map.of(), java.util.Map.of());
+        var context = new TransferContext(UUID.randomUUID(), PLAYER.identity(), source, target, 1, 2, 3, 4,
+                Instant.now());
+        var preparing = new TransferPreparingEvent(context);
+        var released = new SourceReleasedEvent(context);
+        assertEquals(context, preparing.context());
+        assertEquals(context, released.context());
+        assertEquals(new TransferDecision.Allowed(), TransferDecision.allow());
+        assertEquals("busy", ((TransferDecision.Denied) TransferDecision.deny("busy")).reason());
+        SourceReleasedHandler callback = event -> CompletableFuture.completedFuture(null);
+        assertEquals(callback, ((TransferDecision.ReleaseSource) TransferDecision.releaseSource(callback)).handler());
+        assertThrows(IllegalArgumentException.class, () -> TransferDecision.deny(" "));
+        assertThrows(IllegalArgumentException.class, () -> new TransferContext(
+                UUID.randomUUID(), PLAYER.identity(), source, target, 0, 2, 3, 4, Instant.now()));
+    }
+
 }

smoke/UraniumTransferPlugin.java

@@ -9,6 +9,8 @@ import dev.moonbridge.api.PluginContext;
 import dev.moonbridge.api.ServerDefinition;
 import dev.moonbridge.api.ServerRegistration;
 import dev.moonbridge.api.TransferStatus;
+import dev.moonbridge.api.event.TransferDecision;
+import dev.moonbridge.api.event.TransferPreparingEvent;
 
 import java.net.URI;
 import java.util.Map;
@@ -31,6 +33,28 @@ public final class UraniumTransferPlugin implements Plugin {
             throw new IllegalArgumentException("returnToOld must be true or false");
         }
         returnToOld = Boolean.parseBoolean(returnSetting);
+        String releaseSetting = context.settings().getOrDefault("releaseSource", "false");
+        if (!releaseSetting.equals("true") && !releaseSetting.equals("false")) {
+            throw new IllegalArgumentException("releaseSource must be true or false");
+        }
+        if (Boolean.parseBoolean(releaseSetting)) {
+            context.events().subscribe(TransferPreparingEvent.class, preparing -> {
+                context.logger().info("SMOKE_PREPARATION_PASS source={} target={} transfer={}",
+                        preparing.context().source().name(), preparing.context().target().name(),
+                        preparing.context().transferId());
+                return CompletableFuture.completedFuture(TransferDecision.releaseSource(released -> {
+                    PlayerView waiting = context.players().find(released.context().player()).orElseThrow();
+                    if (waiting.currentServer().isPresent()) {
+                        return CompletableFuture.failedFuture(new AssertionError("released source still current"));
+                    }
+                    context.logger().info("SMOKE_SOURCE_RELEASED_PASS transfer={}", released.context().transferId());
+                    // Exercise a genuinely asynchronous barrier; this smoke does not certify a database save.
+                    return CompletableFuture.runAsync(() -> context.logger().info(
+                            "SMOKE_SOURCE_CONFIRMATION_PASS transfer={}", released.context().transferId()),
+                            CompletableFuture.delayedExecutor(250, TimeUnit.MILLISECONDS));
+                }));
+            });
+        }
         targetRegistration = context.servers().register(new ServerDefinition("new",
                 URI.create("tcp://127.0.0.1:" + port), Map.of(),
                 Map.of("source", "uranium-smoke-plugin")));

smoke/local-uranium-transfer.ps1

@@ -2,7 +2,9 @@
 param(
     [Parameter(Mandatory)][string]$BundlePath,
     [string]$Java8Path = 'C:\Program Files\Zulu\zulu-8\bin\java.exe',
+    [string]$BackendJavaAgent,
     [switch]$InstalledPlugin,
+    [switch]$ReleaseSource,
     [switch]$DebugSession,
     [switch]$TraceBackend,
     [switch]$PrismClient,
@@ -13,6 +15,8 @@ param(
 )
 
 $ErrorActionPreference = 'Stop'
+if ($ReleaseSource -and -not $InstalledPlugin) { throw '-ReleaseSource requires -InstalledPlugin' }
+if ($BackendJavaAgent) { $BackendJavaAgent = (Resolve-Path -LiteralPath $BackendJavaAgent).Path }
 if ($PrismClient -and -not $InstalledPlugin) { throw '-PrismClient requires -InstalledPlugin' }
 if ($DebugSession -and -not $InstalledPlugin) { throw '-DebugSession requires -InstalledPlugin' }
 if ($TraceBackend -and -not $InstalledPlugin) { throw '-TraceBackend requires -InstalledPlugin' }
@@ -128,6 +132,9 @@ motd=MoonBridge Uranium transfer smoke
     $info.RedirectStandardInput = $true
     $info.RedirectStandardOutput = $true
     $info.RedirectStandardError = $true
+    if ($BackendJavaAgent) {
+        $info.ArgumentList.Add("-javaagent:${BackendJavaAgent}=$(Join-Path $directory 'class-dump')")
+    }
     foreach ($argument in @('-Xms512m', '-Xmx1024m', '-jar', $serverJar[0].Name, 'nogui')) {
         $info.ArgumentList.Add($argument)
     }
@@ -207,6 +214,7 @@ try {
         $config = Join-Path $runDir 'moonbridge.yml'
         $pluginDirYaml = $pluginDir.Replace('\', '/')
         $returnToOldSetting = if ($ReturnToOld) { 'true' } else { 'false' }
+        $releaseSourceSetting = if ($ReleaseSource) { 'true' } else { 'false' }
         @"
 listen: "127.0.0.1:$proxyPort"
 authentication: OFFLINE
@@ -219,6 +227,7 @@ plugins:
     dev.moonbridge.smoke.UraniumTransferPlugin:
       newPort: "$newRoutePort"
       returnToOld: "$returnToOldSetting"
+      releaseSource: "$releaseSourceSetting"
 backends:
   - name: old
     address: "127.0.0.1:$oldRoutePort"
@@ -328,6 +337,12 @@ backends:
         WaitForProxyLog $proxyLog 'SMOKE_PLUGIN_REGISTER_PASS name=new'
         WaitForProxyLog $proxyLog 'SMOKE_PLUGIN_TRANSFER_PASS status=NETWORK_READY'
         if ($ReturnToOld) { WaitForProxyLog $proxyLog 'SMOKE_PLUGIN_RETURN_PASS status=NETWORK_READY' }
+        if ($ReleaseSource) {
+            WaitForProxyLog $proxyLog 'SMOKE_PREPARATION_PASS'
+            WaitForProxyLog $proxyLog 'SMOKE_SOURCE_RELEASED_PASS'
+            WaitForProxyLog $proxyLog 'SMOKE_SOURCE_CONFIRMATION_PASS'
+            Write-Output 'REAL_URANIUM_SOURCE_RELEASE_PASS asyncBarrier=true persistenceVerified=false'
+        }
     } else {
         & java -cp "$classes;$classpath" dev.moonbridge.smoke.UraniumTransferProbe $oldPort $newPort
         if ($LASTEXITCODE -ne 0) { throw "Uranium transfer probe failed: $LASTEXITCODE" }

smoke/results/2026-10-01-transfer-coordination.md

@@ -0,0 +1,44 @@
+# Transfer coordination validation — 2026-10-01
+
+Baseline: `main@876de1a056a79b807a03f3877cbe6414b60e86dc`. Contract: [transfer coordination](../../docs/transfer-coordination.md). This record covers the optional generic proxy contract, not PlayerDataSQL persistence or mod inventory synchronization.
+
+## Automated checks
+
+On Windows with Zulu OpenJDK 25.0.4.1, `./gradlew.bat --no-daemon check :proxy-core:installDist` succeeded in 6 minutes 26 seconds. The generated JUnit reports contained 49 suites and 309 tests, with zero failures, errors or skips. The build also passed installed-distribution help/version/configuration checks, backend API/host artifact checks, and the native LuckPerms permission acceptance run.
+
+The focused transfer suite had 29 passing cases across `SourceReleaseTransferTest`, `SessionTransferTest`, and `TransferFrameBufferTest`. Coverage includes source EOF before confirmation, confirmation before target LoginStart, a published player with no current server during the gap, discarded gap frames, preparation failure preserving the source relay, incomplete-frame rollback and replay, candidate death before/after source release, frontend cancellation, late callbacks, target registration replacement, buffer ownership/release, and existing vanilla/Forge transitions.
+
+Two concrete defects were found and repaired before this result:
+
+- Buffers must retain frames while source detach can still be rolled back. They switch to discard mode only after detach becomes irreversible, so failed preparation/cutover cannot silently lose retained source operations.
+- A ready candidate must retain same-read-batch packets until its handler is actually removed. `handOff()` now executes immediately before pipeline removal on the same event loop. The regression test sends JoinGame and Position/Look together and checks Position/Look reaches the client.
+
+## Real Uranium round trip
+
+The installed proxy distribution loaded an external SPI smoke plugin and dynamically registered its destination. The plugin subscribed through the public `TransferPreparingEvent` API and requested source release on both transfers. Its confirmation callback completed asynchronously after a 250 ms delay. Two isolated offline Uranium servers ran on Java 8.0.492, with Forge 10.13.4.1614 and no additional gameplay mods or PlayerDataSQL plugins.
+
+Server artifact: `Uranium-1710-rfg-bridge-d7a68f934f-dirty-server.jar`, SHA-256 `A05ECB7CA50CB00390A690608CC243738711C2EC079C37FBA3B0119BB261373A`. A read-only instrumentation agent observed selected final-defined classes to investigate the later backend data-load design; it returned null from its transformer and did not replace class bytes. This run is not a performance measurement.
+
+Command:
+
+```powershell
+./smoke/local-uranium-transfer.ps1 -BundlePath build/uranium-event-bundle -InstalledPlugin -ReleaseSource -ReturnToOld -DebugSession -BackendJavaAgent <observational-agent.jar>
+```
+
+The protocol-5 client completed Forge A→B→A. Each direction received one Forge reset, one replacement ServerHello and two world-transition Respawns; after returning to A the client sustained two KeepAlives. Both transfer futures returned `NETWORK_READY`. Proxy callback logs demonstrate off-I/O execution:
+
+| Forward transfer step | Local timestamp | Thread |
+| --- | --- | --- |
+| Preparation | 02:25:02.639 | moonbridge-plugin-transfer-1 |
+| Source socket released | 02:25:02.644 | moonbridge-plugin-transfer-1 |
+| Asynchronous confirmation | 02:25:02.897 | ForkJoinPool.commonPool-worker-1 |
+| Destination login frames flushed | 02:25:02.899 | moonbridge-session-io-2 |
+| Transfer ready | 02:25:03.124 | uranium-transfer-smoke |
+
+The return confirmation was at 02:25:05.378; destination login frames followed at 02:25:05.379 and transfer completion at 02:25:05.484. The script verified both backend login/disconnect logs and stopped its own proxy and servers. The existing user demo was not modified.
+
+Local raw evidence: `build/local-uranium-transfer-66c44a7f4ab4411a88ea09f4cc1fa231/`, including `proxy.stdout.log`, backend logs and final-defined class dumps. These disposable runtime files are excluded from Git.
+
+## Limits
+
+The synthetic protocol client is not a real Prism visual/gameplay test. The smoke callback confirms only its deliberate asynchronous barrier; it does not observe a database commit. Additional mod inventories, a production Uranium build, arbitrary plugin behavior and PlayerDataSQL crash recovery are outside this result. Performance evidence and its uncertainty are recorded separately in [the benchmark report](../../benchmarks/results/transfer-coordination-validation.md).