fix(anvil): finalize queued dispatch ownership safely

README.md

@@ -28,8 +28,7 @@ world, or accept the EULA. The native suite executes real `ServerPlaceRecipe`, `
 known/highlight codec persistence, and `ResultSlot` ingredient consumption for both Sorting Stick variants and the alternative Dispenser recipe, including repeated batch completion.
 The narrowly scoped anvil-generation adapter also uses this checksum-pinned native runtime as a
 compile-only dependency. Server/API and ProtocolLib classes are not bundled.
-CraftServer, CraftPlayer, and client boundaries are doubled, so this is not evidence of live client
-rendering, search interaction, packet transport, or a full server lifecycle. Those remain
+CraftServer, CraftPlayer, and client boundaries are doubled; byte I/O is a controlled port. Anvil tests execute actual ProtocolLib forwarding, one registered local Netty pipeline/event loop, the declared library scheduler with a queued main-task port, and the real native packet queue, including final cancellation/replacement, skipped handling, owner disable and late-forward retirement. This is not evidence of live client rendering, search interaction, socket/installed-stack packet transport, or a full server lifecycle. Those remain
 non-blocking live checks.
 
 ## Sorting Stick recipe
@@ -105,7 +104,7 @@ Default risk uses the **largest resulting enchantment excess above its normal ma
 | 6 | 95% |
 | 7 or more | 99% |
 
-ProtocolLib binds taking to the displayed menu generation before payment. A bounded 40-server-tick admission accommodates the main → Netty → native-main handoff; an owned native dispatch shim carries exact packet identity through that queue, so one click cannot borrow another click's cancellation or payment ticket. Native aggregate and packet-specific limits remain authoritative, using the actual connection's counters and violation actions; expired/refused/cancelled dispatches never pay. Input or acknowledged risk changes invalidate old clicks even when the result item is identical; the ordinary anvil stays open without another confirmation. The adapter is pinned to Purpur 26.2 build 2618 and fails closed when enhanced generation admission is unavailable. This is a library dependency, not a Fantasy Stick/gameplay-plugin bridge.
+ProtocolLib binds taking to the displayed menu generation before payment. A bounded 40-server-tick admission accommodates the main → Netty → native-main handoff; an owned native dispatch shim carries exact packet identity through that queue, so one click cannot borrow another click's cancellation or payment ticket. Native aggregate and packet-specific limits remain authoritative, using the actual connection's counters and violation actions; expired/refused/cancelled dispatches never pay. Transport ownership is bounded to 32 records per connection. A queued ProtocolLib-owned main callback followed by a barrier on the same native channel event loop proves whether each exact raw was forwarded; mutable packet/post-event snapshots do not. This library-owned cleanup survives this plugin's task cancellation. Revoked native wrappers retain their denial even when native admission or queue processing skips their handler, while genuinely unresolved late raws keep protective ownership until arrival, the ordered fence or disconnect. Missing/rejected library scheduling or event-loop execution fails closed; it does not establish successful cleanup. Input or acknowledged risk changes invalidate old clicks even when the result item is identical; the ordinary anvil stays open without another confirmation. The adapter is pinned to Purpur 26.2 build 2618 and fails closed when enhanced generation admission is unavailable. This is a library dependency, not a Fantasy Stick/gameplay-plugin bridge.
 
 Use left/right pickup, shift-click or a native-eligible empty hotbar/offhand destination. Full destinations, cancelled/stale takes and unsupported cloning/drop/collection actions do not roll or spend items. Ordinary anvil wear remains in effect. Dedicated gameplay-plugin compatibility is unsupported; declared native cancellations and unrelated inventory state are preserved.
 

src/main/java/games/dmg/spigotinventoryhelper/AdvancedAnvilProtocolGuard.java

@@ -33,7 +33,8 @@ final class AdvancedAnvilProtocolGuard implements AutoCloseable {
         BukkitTask expiry;
         volatile boolean sealed;
         volatile boolean revoked;
-        volatile boolean handedOff;
+        // Transport-monitor protected, sticky raw-seen/wrapper ownership (not post acknowledgement).
+        boolean wrapped;
         boolean consumed;
         final AnvilView view;
         final net.minecraft.server.network.ServerGamePacketListenerImpl nativeListener;
@@ -121,7 +122,7 @@ final class AdvancedAnvilProtocolGuard implements AutoCloseable {
                         nativePlayer.level().getServer().packetProcessor(), () -> {
                             if (!closed && !plugin.getServer().isPrimaryThread()) plugin.getServer().getScheduler().runTask(plugin,
                                     () -> { if (!closed && !watched.contains(view)) owner.protocolHookReady(view); });
-                        });
+                        }, () -> plugin.getServer().isPrimaryThread());
                 dispatches.put(player, dispatch);
             }
             if (!dispatch.ready()) return false;
@@ -192,6 +193,8 @@ final class AdvancedAnvilProtocolGuard implements AutoCloseable {
         }
         if (!(event.getPlayer().getOpenInventory() instanceof AnvilView view) || !watched.contains(view)) return;
         invalidate(view); // A new raw intent can never reuse an earlier ticket.
+        NativeAnvilDispatch acquiredDispatch = null;
+        Admission acquired = null;
         try {
             Object raw = event.getPacket().getHandle();
             if (seen.put(event, Boolean.TRUE) != null || event.isCancelled()
@@ -223,27 +226,30 @@ final class AdvancedAnvilProtocolGuard implements AutoCloseable {
                 if (!admission.consumed && !closed && !admissions.containsKey(view)) owner.protocolRejected(view);
             };
             if (!dispatch.admit(admission)) { refuse(event, view); return; }
+            acquiredDispatch = dispatch;
+            acquired = admission;
             admissions.put(view, admission);
-            var marker = com.comphenix.protocol.events.NetworkMarker.getNetworkMarker(event);
-            if (marker == null) {
-                marker = new com.comphenix.protocol.events.NetworkMarker(
-                        com.comphenix.protocol.events.ConnectionSide.CLIENT_SIDE, event.getPacketType());
-                event.setNetworkMarker(marker);
-            }
-            marker.addPostListener(new com.comphenix.protocol.events.PacketPostListener() {
-                @Override public Plugin getPlugin() { return plugin; }
-                @Override public void onPostEvent(PacketEvent forwarded) {
-                    if (forwarded == event) admission.handedOff = true;
-                }
-            });
-            admission.expiry = plugin.getServer().getScheduler().runTaskLater(plugin, () -> {
-                admission.revoked = true;
+            // No owned post listener/marker: foreign markers remain entirely untouched.
+            admission.expiry = Objects.requireNonNull(plugin.getServer().getScheduler().runTaskLater(plugin, () -> {
+                dispatch.revoke(admission);
                 if (admissions.get(view) == admission) admissions.remove(view);
-                // Post means forward scheduled, not native completion. Keep such tombstones
-                // until the exact record reaches the hook or its queued wrapper finishes.
-                if (!admission.handedOff) dispatch.complete(admission);
-            }, ADMISSION_TICKS);
-        } catch (RuntimeException | LinkageError unavailable) { refuse(event, view); }
+                // Unresolved raw ownership waits for the two-queue fence or exact raw arrival.
+                // Already-wrapped ownership is safe to release even if native skips handle.
+            }, ADMISSION_TICKS));
+        } catch (RuntimeException | LinkageError unavailable) {
+            if (acquired != null) {
+                acquiredDispatch.revoke(acquired);
+                if (admissions.get(view) == acquired) admissions.remove(view);
+            }
+            refuse(event, view);
+        } finally {
+            // Also fences a partially acquired, cancelled record after timer setup throws.
+            // Cancellation inside a callback is never upfront proof of no raw forward.
+            if (acquired != null && !acquiredDispatch.requestFence()) {
+                invalidate(view);
+                refuse(event, view);
+            }
+        }
     }
 
     private boolean currentFrame(AnvilView view, ServerboundContainerClickPacket packet) {

src/main/java/games/dmg/spigotinventoryhelper/NativeAnvilDispatch.java

@@ -24,13 +24,22 @@ final class NativeAnvilDispatch extends ChannelInboundHandlerAdapter {
     private final Runnable readyCallback;
     private final Map<ServerboundContainerClickPacket, AdvancedAnvilProtocolGuard.Admission> records =
             new IdentityHashMap<>();
-    private final java.util.Set<ServerboundContainerClickPacket> dispatched =
-            java.util.Collections.newSetFromMap(new IdentityHashMap<>());
+    private final com.comphenix.protocol.scheduler.ProtocolScheduler scheduler;
+    private final java.util.function.BooleanSupplier mainThread;
+    private boolean mainFencePending;
+    private volatile boolean unavailable;
     private volatile boolean ready;
     private volatile boolean removing;
     private boolean retiring;
 
-    NativeAnvilDispatch(Connection connection, PacketProcessor processor, Runnable readyCallback) {
+    NativeAnvilDispatch(Connection connection, PacketProcessor processor, Runnable readyCallback,
+            java.util.function.BooleanSupplier mainThread) {
+        // Link the public, library-owned main queue BEFORE enhanced output/hook installation.
+        // DefaultScheduler.runTask runs after the executing library filters/posts return;
+        // this is an ordering fence, not a promise to wait a strict server tick.
+        scheduler = java.util.Objects.requireNonNull(com.comphenix.protocol.ProtocolLibrary.getScheduler(),
+                "Missing ProtocolLib main scheduler");
+        this.mainThread = java.util.Objects.requireNonNull(mainThread);
         NativeAnvilPacketLimits.link();
         this.connection = java.util.Objects.requireNonNull(connection);
         this.processor = java.util.Objects.requireNonNull(processor);
@@ -49,7 +58,7 @@ final class NativeAnvilDispatch extends ChannelInboundHandlerAdapter {
     boolean ready() {
         var pipeline = connection.channel.pipeline();
         var context = pipeline.context(this);
-        return ready && !removing && connection.channel.isActive() && context != null
+        return ready && !unavailable && !removing && connection.channel.isActive() && context != null
                 && pipeline.context(connection) != null && pipeline.last() == connection
                 && pipeline.names().indexOf(context.name()) + 1
                     == pipeline.names().indexOf(pipeline.context(connection).name());
@@ -72,13 +81,76 @@ final class NativeAnvilDispatch extends ChannelInboundHandlerAdapter {
         return true;
     }
 
+    /** One pending library-main callback per dispatch, never one job per admission. */
+    synchronized boolean requestFence() {
+        if (unavailable) return false;
+        if (mainFencePending) return true;
+        mainFencePending = true;
+        try {
+            if (scheduler.runTask(this::mainFence) == null) {
+                failBridge();
+                return false;
+            }
+            return true;
+        } catch (RuntimeException | LinkageError failure) {
+            failBridge();
+            return false;
+        }
+    }
+
+    private void mainFence() {
+        try {
+            if (!mainThread.getAsBoolean()) {
+                failBridge();
+                return;
+            }
+            final java.util.List<AdvancedAnvilProtocolGuard.Admission> snapshot;
+            synchronized (this) {
+                mainFencePending = false;
+                if (unavailable) return;
+                // Snapshot HERE, not on Netty: later unfinished receives cannot be declared
+                // NO_FORWARD by a barrier queued before their actual library raw forwards.
+                snapshot = java.util.List.copyOf(records.values()); // bounded by MAX_OUTSTANDING
+            }
+            connection.channel.eventLoop().execute(() -> terminalFence(snapshot));
+        } catch (RuntimeException | LinkageError failure) {
+            // Rejected submission is not evidence of NO_FORWARD.
+            failBridge();
+        }
+    }
+
+    private synchronized void terminalFence(java.util.List<AdvancedAnvilProtocolGuard.Admission> snapshot) {
+        for (var admission : snapshot) {
+            if (records.get(admission.packet) != admission) continue;
+            if (!admission.wrapped) {
+                // All actual raw forwards captured before mainFence are ahead of us on
+                // this same event loop. Never infer acknowledgement from mutable events/posts.
+                admission.revoked = true;
+                records.remove(admission.packet);
+            } else if (admission.revoked) records.remove(admission.packet);
+        }
+        // Transport-only completion: no Bukkit/map/policy callback on the socket loop.
+        removeIfDrained();
+    }
+
+    private synchronized void failBridge() {
+        unavailable = true;
+        mainFencePending = false;
+        revokeRecords(); // Keep unresolved exact-raw tombstones; no false cleanup claim.
+        removeIfDrained();
+    }
+
     static AdvancedAnvilProtocolGuard.Admission current() { return CURRENT.get(); }
 
     @Override public void channelRead(ChannelHandlerContext context, Object message) {
         AdvancedAnvilProtocolGuard.Admission admission;
         synchronized (this) {
             admission = records.get(message);
-            if (admission != null && !dispatched.add(admission.packet)) return;
+            if (admission != null) {
+                if (admission.wrapped) return; // Duplicate exact raw while still owned.
+                admission.wrapped = true; // Sticky, even after ledger retirement.
+                if (admission.revoked) complete(admission);
+            }
         }
         if (admission == null) {
             context.fireChannelRead(message);
@@ -92,13 +164,25 @@ final class NativeAnvilDispatch extends ChannelInboundHandlerAdapter {
 
     synchronized void complete(AdvancedAnvilProtocolGuard.Admission admission) {
         if (records.get(admission.packet) == admission) records.remove(admission.packet);
-        dispatched.remove(admission.packet);
         removeIfDrained();
     }
 
+    synchronized void revoke(AdvancedAnvilProtocolGuard.Admission admission) {
+        admission.revoked = true;
+        // The wrapper retains irreversible denial even if native never calls its handle.
+        if (admission.wrapped) complete(admission);
+    }
+
+    private void revokeRecords() {
+        records.values().removeIf(admission -> {
+            admission.revoked = true;
+            return admission.wrapped;
+        });
+    }
+
     synchronized void retire() {
         retiring = true;
-        records.values().forEach(admission -> admission.revoked = true);
+        revokeRecords();
         removeIfDrained();
     }
 
@@ -116,7 +200,6 @@ final class NativeAnvilDispatch extends ChannelInboundHandlerAdapter {
         synchronized (this) {
             records.values().forEach(admission -> admission.revoked = true);
             records.clear();
-            dispatched.clear();
             retiring = true;
             removeIfDrained();
         }

src/nativeTest/java/games/dmg/spigotinventoryhelper/NativeAnvilCombiningRegressionTest.java

@@ -46,10 +46,11 @@ class NativeAnvilCombiningRegressionTest {
 
     @Test void queuedDeclaredLibraryTakeSurvivesSchedulerBeforeNativeDispatch() throws Exception {
         var rolls = new AtomicInteger();
-        try (var f = new NativeAnvilFixture(directory, () -> { rolls.incrementAndGet(); return 99; }, () -> 1);
+        try (var f = new NativeAnvilFixture(directory, () -> { rolls.incrementAndGet(); return 99; }, () -> 1, true);
                 var packet = new NativeAnvilPacketFixture(f);
                 var transport = new NativeAnvilTransportFixture(f, packet)) {
             f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
             assertEquals(6, f.output().getEnchantmentLevel(Enchantment.SHARPNESS));
             var raw = new net.minecraft.network.protocol.game.ServerboundContainerClickPacket(
                     f.menu.containerId, f.clientState(f.menu), (short) 2, (byte) 0, ContainerInput.PICKUP,
@@ -78,10 +79,11 @@ class NativeAnvilCombiningRegressionTest {
 
     @Test void earlierForwardedCancellationCannotBorrowALaterIdenticalAdmission() throws Exception {
         var rolls = new AtomicInteger();
-        try (var f = new NativeAnvilFixture(directory, () -> { rolls.incrementAndGet(); return 99; }, () -> 1);
+        try (var f = new NativeAnvilFixture(directory, () -> { rolls.incrementAndGet(); return 99; }, () -> 1, true);
                 var packet = new NativeAnvilPacketFixture(f);
                 var transport = new NativeAnvilTransportFixture(f, packet)) {
             f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
             var raw = new net.minecraft.network.protocol.game.ServerboundContainerClickPacket(
                     f.menu.containerId, f.clientState(f.menu), (short) 2, (byte) 0, ContainerInput.PICKUP,
                     new it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap<>(), net.minecraft.network.HashedStack.EMPTY);
@@ -314,10 +316,11 @@ class NativeAnvilCombiningRegressionTest {
     @org.junit.jupiter.params.provider.ValueSource(strings = {"expiry", "inputs", "policy", "close", "shutdown", "native-refusal"})
     void queuedRetiredOrRefusedIntentsNeverDebitAndCannotAuthorizeLaterTakes(String mode) throws Exception {
         var rolls = new AtomicInteger();
-        try (var f = new NativeAnvilFixture(directory, () -> { rolls.incrementAndGet(); return 99; }, () -> 1);
+        try (var f = new NativeAnvilFixture(directory, () -> { rolls.incrementAndGet(); return 99; }, () -> 1, true);
                 var packet = new NativeAnvilPacketFixture(f);
                 var transport = new NativeAnvilTransportFixture(f, packet)) {
             f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
             transport.receive(raw(f));
             transport.forwardToNativeQueue();
             switch (mode) {
@@ -353,11 +356,12 @@ class NativeAnvilCombiningRegressionTest {
     @org.junit.jupiter.params.provider.ValueSource(booleans = {false, true})
     void alreadyQueuedOldIdentityCannotTakeFromAReusedNativeWindow(boolean newAdvanced) throws Exception {
         var rolls = new AtomicInteger();
-        try (var f = new NativeAnvilFixture(directory, () -> { rolls.incrementAndGet(); return 99; }, () -> 1);
+        try (var f = new NativeAnvilFixture(directory, () -> { rolls.incrementAndGet(); return 99; }, () -> 1, true);
                 var packet = new NativeAnvilPacketFixture(f);
                 var transport = new NativeAnvilTransportFixture(f, packet)) {
             nativeExperienceAndWear(f, 100);
             f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
             transport.receive(raw(f)); transport.forwardToNativeQueue();
             packet.closeMenu(org.bukkit.event.inventory.InventoryCloseEvent.Reason.PLAYER);
             var reopened = new net.minecraft.world.inventory.AnvilMenu(f.menu.containerId, f.realInventory,

src/nativeTest/java/games/dmg/spigotinventoryhelper/NativeAnvilFixture.java

@@ -61,7 +61,7 @@ final class NativeAnvilFixture implements AutoCloseable {
     final net.minecraft.network.Connection connection;
     final net.minecraft.server.network.ServerGamePacketListenerImpl transport =
             mock(net.minecraft.server.network.ServerGamePacketListenerImpl.class);
-    final io.netty.channel.embedded.EmbeddedChannel channel;
+    io.netty.channel.Channel channel;
     private java.util.function.Consumer<net.minecraft.network.protocol.game.ServerboundContainerClickPacket> legacyClick;
     final List<Object> unrelatedInbound = new ArrayList<>();
     private final io.papermc.paper.configuration.GlobalConfiguration previousConfiguration;
@@ -79,6 +79,13 @@ final class NativeAnvilFixture implements AutoCloseable {
     final org.bukkit.command.ConsoleCommandSender console = mock(org.bukkit.command.ConsoleCommandSender.class);
     final java.util.concurrent.BlockingQueue<Runnable> nextTick = new java.util.concurrent.LinkedBlockingQueue<>();
     private long tickCount;
+    RuntimeException admissionSchedulingFailure;
+    RuntimeException librarySchedulingFailure;
+    final java.util.Queue<Runnable> libraryTasks = new java.util.concurrent.ConcurrentLinkedQueue<>();
+    final org.bukkit.plugin.Plugin libraryPlugin = mock(org.bukkit.plugin.Plugin.class);
+    private final Object previousLibraryScheduler;
+    private final Object previousLibraryPlugin;
+    private int taskId;
     private final Map<Long, java.util.Collection<Runnable>> delayed = new java.util.TreeMap<>();
     final List<Runnable> repeating = new ArrayList<>();
     final List<BukkitTask> tasks = new ArrayList<>();
@@ -99,8 +106,22 @@ final class NativeAnvilFixture implements AutoCloseable {
         this(directory, risk, wear, "Purpur 26.2.build.2618-stable");
     }
 
+    NativeAnvilFixture(Path directory, boolean queuedTransport) throws Exception {
+        this(directory, () -> 99, () -> 1.0, "Purpur 26.2.build.2618-stable", queuedTransport);
+    }
+
+    NativeAnvilFixture(Path directory, java.util.function.IntSupplier risk,
+            java.util.function.DoubleSupplier wear, boolean queuedTransport) throws Exception {
+        this(directory, risk, wear, "Purpur 26.2.build.2618-stable", queuedTransport);
+    }
+
     NativeAnvilFixture(Path directory, java.util.function.IntSupplier risk,
             java.util.function.DoubleSupplier wear, String version) throws Exception {
+        this(directory, risk, wear, version, false);
+    }
+
+    private NativeAnvilFixture(Path directory, java.util.function.IntSupplier risk,
+            java.util.function.DoubleSupplier wear, String version, boolean queuedTransport) throws Exception {
         NativeRuntime.bootstrap();
         previousConfiguration = io.papermc.paper.configuration.GlobalConfiguration.get();
         if (previousConfiguration == null) {
@@ -128,7 +149,7 @@ final class NativeAnvilFixture implements AutoCloseable {
             }
             return null;
         }).when(connection).channelRead(any(), any());
-        channel = new io.netty.channel.embedded.EmbeddedChannel();
+        channel = queuedTransport ? new NativeAnvilTestChannel() : new io.netty.channel.embedded.EmbeddedChannel();
         channel.pipeline().addLast("packet_handler", connection);
         connection.channel = channel;
         nativePlayer.connection = transport;
@@ -145,7 +166,8 @@ final class NativeAnvilFixture implements AutoCloseable {
         setServer(server);
         when(server.getItemFactory()).thenAnswer(ignored -> CraftItemFactory.instance());
         when(server.getUnsafe()).thenReturn(CraftMagicNumbers.INSTANCE);
-        when(server.isPrimaryThread()).thenReturn(true);
+        var mainThread = Thread.currentThread();
+        when(server.isPrimaryThread()).thenAnswer(ignored -> Thread.currentThread() == mainThread);
         when(server.getVersion()).thenReturn("Purpur 2618 (MC: 26.2)");
         when(server.getMinecraftVersion()).thenReturn("26.2");
         when(server.getBukkitVersion()).thenReturn("26.2.build.2618-stable");
@@ -166,12 +188,32 @@ final class NativeAnvilFixture implements AutoCloseable {
         doReturn(List.of()).when(server).getOnlinePlayers();
         var scheduler = mock(org.bukkit.craftbukkit.scheduler.CraftScheduler.class);
         when(server.getScheduler()).thenReturn(scheduler);
-        when(scheduler.runTask(any(), any(Runnable.class))).thenAnswer(inv -> schedule(inv.getArgument(1), nextTick));
+        when(scheduler.runTask(any(), any(Runnable.class))).thenAnswer(inv -> {
+            if (inv.getArgument(0) == libraryPlugin) {
+                if (!libraryPlugin.isEnabled()) throw new IllegalStateException("Required library disabled");
+                if (librarySchedulingFailure != null) throw librarySchedulingFailure;
+                return schedule(inv.getArgument(1), libraryTasks);
+            }
+            return schedule(inv.getArgument(1), nextTick);
+        });
+        when(libraryPlugin.getServer()).thenReturn(server);
+        when(libraryPlugin.isEnabled()).thenReturn(true);
+        var schedulerField = com.comphenix.protocol.ProtocolLibrary.class.getDeclaredField("scheduler");
+        schedulerField.setAccessible(true);
+        previousLibraryScheduler = schedulerField.get(null);
+        var pluginField = com.comphenix.protocol.ProtocolLibrary.class.getDeclaredField("plugin");
+        pluginField.setAccessible(true);
+        previousLibraryPlugin = pluginField.get(null);
+        schedulerField.set(null, new com.comphenix.protocol.scheduler.DefaultScheduler(libraryPlugin));
+        pluginField.set(null, libraryPlugin);
         when(scheduler.runTaskLater(any(), any(Runnable.class), anyLong())).thenAnswer(inv -> {
+            if (admissionSchedulingFailure != null) throw admissionSchedulingFailure;
             long delay = inv.getArgument(2);
             if (delay < 1) throw new AssertionError("Unexpected non-positive delayed task");
             return schedule(inv.getArgument(1), delayed.computeIfAbsent(tickCount + delay, ignored -> new ArrayList<>()));
         });
+        doAnswer(ignored -> { List.copyOf(tasks).forEach(BukkitTask::cancel); return null; })
+                .when(scheduler).cancelTasks(any(org.bukkit.plugin.Plugin.class));
         when(scheduler.runTaskTimer(any(), any(Runnable.class), anyLong(), anyLong()))
                 .thenAnswer(inv -> schedule(inv.getArgument(1), repeating));
         when(player.getHandle()).thenReturn(nativePlayer);
@@ -322,7 +364,7 @@ final class NativeAnvilFixture implements AutoCloseable {
 
     void dispatchInbound(Object record) {
         channel.pipeline().fireChannelRead(record);
-        channel.checkException();
+        if (channel instanceof io.netty.channel.embedded.EmbeddedChannel embedded) embedded.checkException();
     }
 
     private void dispatchPacket(net.minecraft.network.protocol.Packet<?> raw) {
@@ -389,7 +431,15 @@ final class NativeAnvilFixture implements AutoCloseable {
         return menu.getBukkitView().getTopInventory().getItem(2);
     }
 
+    void pumpLibraryTasks() {
+        if (!server.isPrimaryThread()) throw new AssertionError("Library callbacks must run on main");
+        Runnable task;
+        while ((task = libraryTasks.poll()) != null) task.run();
+        if (channel instanceof io.netty.channel.embedded.EmbeddedChannel embedded) embedded.runPendingTasks();
+    }
+
     void tick() {
+        pumpLibraryTasks();
         var due = delayed.remove(++tickCount);
         if (due != null) List.copyOf(due).forEach(Runnable::run);
         var ready = List.copyOf(nextTick);
@@ -399,6 +449,13 @@ final class NativeAnvilFixture implements AutoCloseable {
         assertNoEventErrors();
     }
 
+    // Paper disables the plugin before cancelling its scheduler tasks. Do not let a fixture
+    // silently supply the expiry callbacks that real disable removes.
+    void disableAndCancelTasks() {
+        plugin.setEnabled(false);
+        server.getScheduler().cancelTasks(plugin);
+    }
+
     void assertNoEventErrors() {
         if (!eventErrors.isEmpty()) throw new AssertionError("Native listener failed", eventErrors.getFirst());
     }
@@ -407,7 +464,9 @@ final class NativeAnvilFixture implements AutoCloseable {
         queue.add(runnable);
         BukkitTask task = mock(BukkitTask.class);
         doAnswer(ignored -> { queue.remove(runnable); return null; }).when(task).cancel();
-        tasks.add(task);
+        when(task.getTaskId()).thenReturn(++taskId);
+        // Only owning-plugin tasks belong to realistic owner-disable cancellation.
+        if (queue != libraryTasks) tasks.add(task);
         return task;
     }
 
@@ -415,7 +474,10 @@ final class NativeAnvilFixture implements AutoCloseable {
         try {
             plugin.setEnabled(false);
             HandlerList.unregisterAll(plugin);
-            channel.finishAndReleaseAll();
+            if (channel instanceof io.netty.channel.embedded.EmbeddedChannel embedded) embedded.finishAndReleaseAll();
+            else ((NativeAnvilTestChannel) channel).finish();
+            field(com.comphenix.protocol.ProtocolLibrary.class, null, "scheduler", previousLibraryScheduler);
+            field(com.comphenix.protocol.ProtocolLibrary.class, null, "plugin", previousLibraryPlugin);
             processor.close();
             field(io.papermc.paper.configuration.GlobalConfiguration.class, null, "instance", previousConfiguration);
             setServer(previous);

src/nativeTest/java/games/dmg/spigotinventoryhelper/NativeAnvilLifecycleRegressionTest.java

@@ -0,0 +1,342 @@
+package games.dmg.spigotinventoryhelper;
+
+import static games.dmg.spigotinventoryhelper.NativeAnvilCombiningRegressionTest.raw;
+import static games.dmg.spigotinventoryhelper.NativeAnvilCombiningRegressionTest.sword;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+import com.comphenix.protocol.PacketType;
+import com.comphenix.protocol.events.*;
+import java.nio.file.Path;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import net.minecraft.network.HashedStack;
+import net.minecraft.network.protocol.game.ServerboundContainerClickPacket;
+import net.minecraft.world.inventory.ContainerInput;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/** Terminal ownership through actual local ProtocolLib forwarding/pipeline/native queue.
+ * No sockets, live server, or arbitrary gameplay-plugin compatibility claim.
+ */
+class NativeAnvilLifecycleRegressionTest {
+    @TempDir Path directory;
+
+    @ParameterizedTest @ValueSource(strings = {"replace", "cancel"})
+    void finalFilterNoForwardOfAdmittedHandleRestoresCapacityAtExpiry(String mode) throws Exception {
+        try (var f = new NativeAnvilFixture(directory, true);
+                var packet = new NativeAnvilPacketFixture(f);
+                var transport = new NativeAnvilTransportFixture(f, packet)) {
+            f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
+            var foreign = filter(f, event -> {
+                if (mode.equals("cancel")) event.setCancelled(true);
+                else event.setPacket(new PacketContainer(event.getPacketType(), inert(f)));
+            });
+            for (int i = 0; i < NativeAnvilDispatch.MAX_OUTSTANDING; i++) transport.receive(raw(f));
+            for (var event : transport.received) assertEquals(mode.equals("cancel"), event.isCancelled());
+            expire(f);
+            transport.forwardToNativeQueue(); // Resolve queued fence before asserting capacity.
+            f.protocol.removePacketListener(foreign);
+            transport.receive(raw(f));
+            assertFalse(transport.received.getLast().isCancelled(),
+                    "Final filter never forwarded admitted A: expiry must restore its capacity");
+            assertTrue(packet.clicks.isEmpty());
+        }
+    }
+
+    @Test void cancelledPacketDisableDoesNotRelyOnPluginExpiryAndPreservesForeignHandler() throws Exception {
+        try (var f = new NativeAnvilFixture(directory, true);
+                var packet = new NativeAnvilPacketFixture(f);
+                var transport = new NativeAnvilTransportFixture(f, packet)) {
+            var probe = new io.netty.channel.ChannelInboundHandlerAdapter();
+            f.channel.pipeline().addBefore("packet_handler", "foreign-probe", probe);
+            f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
+            filter(f, event -> event.setCancelled(true));
+            int before = f.channel.pipeline().names().size();
+            transport.receive(raw(f));
+            assertTrue(transport.received.getFirst().isCancelled());
+            f.disableAndCancelTasks();
+            expire(f);
+            transport.forwardToNativeQueue();
+            assertFalse(transport.processor.executeSinglePacket(), "Cancelled means no native forward");
+            assertSame(probe, f.channel.pipeline().get("foreign-probe"));
+            assertSame(f.connection, f.channel.pipeline().get("packet_handler"));
+            assertEquals(before - 1, f.channel.pipeline().names().size(),
+                    "Resolved no-forward shutdown removes only owned hook without a plugin timer");
+            assertTrue(packet.clicks.isEmpty());
+        }
+    }
+
+    @Test void admissionTimerAcquisitionFailureRollsBackCapacity() throws Exception {
+        try (var f = new NativeAnvilFixture(directory, true);
+                var packet = new NativeAnvilPacketFixture(f);
+                var transport = new NativeAnvilTransportFixture(f, packet)) {
+            f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
+            f.admissionSchedulingFailure = new IllegalStateException("Controlled timer acquisition failure");
+            for (int i = 0; i < NativeAnvilDispatch.MAX_OUTSTANDING; i++) {
+                transport.receive(raw(f));
+                assertTrue(transport.received.getLast().isCancelled());
+            }
+            transport.forwardToNativeQueue();
+            f.admissionSchedulingFailure = null;
+            transport.receive(raw(f));
+            assertFalse(transport.received.getLast().isCancelled(),
+                    "Failed acquisition must not fill the exact-identity budget");
+            assertTrue(packet.clicks.isEmpty());
+        }
+    }
+
+    @Test void nativeQueueSkipWithoutWrapperHandleRestoresCapacityAtExpiry() throws Exception {
+        try (var f = new NativeAnvilFixture(directory, true);
+                var packet = new NativeAnvilPacketFixture(f);
+                var transport = new NativeAnvilTransportFixture(f, packet)) {
+            f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
+            for (int i = 0; i < NativeAnvilDispatch.MAX_OUTSTANDING; i++) {
+                transport.receive(raw(f));
+                assertFalse(transport.received.getLast().isCancelled());
+            }
+            transport.forwardToNativeQueue();
+            when(packet.handler.shouldHandleMessage(any())).thenReturn(false);
+            for (int i = 0; i < NativeAnvilDispatch.MAX_OUTSTANDING; i++) {
+                assertTrue(transport.processor.executeSinglePacket());
+            }
+            assertFalse(transport.processor.executeSinglePacket());
+            assertTrue(packet.clicks.isEmpty(), "Native queue really skipped wrapper.handle");
+            expire(f);
+            when(packet.handler.shouldHandleMessage(any())).thenReturn(true);
+            transport.receive(raw(f));
+            assertFalse(transport.received.getLast().isCancelled(),
+                    "Revoked already-wrapped ownership can retire without handle/finally");
+        }
+    }
+
+    @Test void aggregateNativeRefusalWithoutWrapperHandleRetiresHookAtExpiryAndClose() throws Exception {
+        try (var f = new NativeAnvilFixture(directory, true);
+                var packet = new NativeAnvilPacketFixture(f);
+                var transport = new NativeAnvilTransportFixture(f, packet)) {
+            var config = io.papermc.paper.configuration.GlobalConfiguration.get();
+            var previous = config.packetLimiter;
+            try {
+                var limiter = config.new PacketLimiter();
+                limiter.allPackets = new io.papermc.paper.configuration.GlobalConfiguration.PacketLimiter.PacketLimit(
+                        1, 0.1, io.papermc.paper.configuration.GlobalConfiguration.PacketLimiter.PacketLimit.ViolateAction.KICK);
+                limiter.overrides = new java.util.HashMap<>();
+                config.packetLimiter = limiter;
+                var connection = new net.minecraft.network.Connection(net.minecraft.network.protocol.PacketFlow.SERVERBOUND);
+                connection.channel = f.channel;
+                NativeAnvilFixture.field(net.minecraft.network.Connection.class, connection, "packetListener", f.transport);
+                NativeAnvilFixture.field(net.minecraft.server.network.ServerCommonPacketListenerImpl.class,
+                        f.transport, "connection", connection);
+                f.channel.pipeline().replace(f.connection, "packet_handler", connection);
+                // Keep the inspected frame-synchronizer port: changing the listener's
+                // connection must not bypass the fixture's actual ProtocolLib frame publication.
+                doAnswer(inv -> { f.connection.send(inv.getArgument(0), inv.getArgument(1), true); return null; })
+                        .when(f.transport).send(any(net.minecraft.network.protocol.Packet.class), any());
+                var disconnects = new java.util.concurrent.CopyOnWriteArrayList<Object>();
+                // Controlled outbound port: hold completion of the real native disconnect
+                // write so the channel stays active; do not silently let disconnect clear ownership.
+                var foreign = new io.netty.channel.ChannelOutboundHandlerAdapter() {
+                    @Override public void write(io.netty.channel.ChannelHandlerContext context, Object message,
+                            io.netty.channel.ChannelPromise promise) { disconnects.add(message); }
+                };
+                f.channel.pipeline().addFirst("held-disconnect-write", foreign);
+                f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
+                int before = f.channel.pipeline().names().size();
+                transport.receive(raw(f));
+                assertFalse(transport.received.getFirst().isCancelled());
+                transport.forwardToNativeQueue();
+                assertFalse(transport.processor.executeSinglePacket(), "Aggregate refusal happens before wrapper.handle");
+                assertTrue(disconnects.stream().anyMatch(message -> message instanceof
+                        net.minecraft.network.protocol.common.ClientboundDisconnectPacket), "Native kick path actually executed");
+                assertTrue(f.channel.isActive(), "Retirement proof must not rely on channelInactive");
+                expire(f);
+                packet.closeMenu(org.bukkit.event.inventory.InventoryCloseEvent.Reason.PLAYER);
+                assertSame(foreign, f.channel.pipeline().get("held-disconnect-write"));
+                assertSame(connection, f.channel.pipeline().get("packet_handler"));
+                assertEquals(before - 1, f.channel.pipeline().names().size(),
+                        "Already-wrapped aggregate refusal must not strand retired owned hook");
+                assertTrue(packet.clicks.isEmpty());
+            } finally { config.packetLimiter = previous; }
+        }
+    }
+
+    @ParameterizedTest @ValueSource(strings = {"filter", "post"})
+    void reentrantDisableKeepsUnresolvedOrScheduledExactRawSafeBeyondExpiryAndWindowReuse(String phase)
+            throws Exception {
+        var rolls = new AtomicInteger();
+        try (var f = new NativeAnvilFixture(directory, () -> { rolls.incrementAndGet(); return 99; }, () -> 1, true);
+                var packet = new NativeAnvilPacketFixture(f);
+                var transport = new NativeAnvilTransportFixture(f, packet)) {
+            NativeAnvilCombiningRegressionTest.nativeExperienceAndWear(f, 100);
+            f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
+            filter(f, event -> {
+                if (phase.equals("filter")) {
+                    // Cancellation is NOT terminal inside a filter: the same callback can
+                    // still restore the captured exact raw before ProtocolLib schedules it.
+                    event.setCancelled(true);
+                    f.disableAndCancelTasks();
+                    event.setCancelled(false);
+                } else {
+                    post(f, event, forwarded -> {
+                        // ProtocolLib has already captured/scheduled the exact raw. A
+                        // post cancellation is not proof that this raw will never arrive.
+                        forwarded.setReadOnly(false);
+                        forwarded.setCancelled(true);
+                        f.disableAndCancelTasks();
+                    });
+                }
+            });
+            transport.receive(raw(f));
+            assertEquals(phase.equals("post"), transport.received.getFirst().isCancelled());
+            expire(f); f.tick(); // Raw exact forward is still held in the real library event loop.
+            var reopened = new net.minecraft.world.inventory.AnvilMenu(f.menu.containerId, f.realInventory,
+                    net.minecraft.world.inventory.ContainerLevelAccess.create(f.level, f.position));
+            reopened.setTitle(net.minecraft.network.chat.Component.literal("Repair & Name"));
+            f.nativePlayer.containerMenu = reopened;
+            reopened.setSynchronizer(packet.synchronizer);
+            reopened.getSlot(0).set(org.bukkit.craftbukkit.inventory.CraftItemStack.asNMSCopy(sword(4)));
+            reopened.getSlot(1).set(org.bukkit.craftbukkit.inventory.CraftItemStack.asNMSCopy(sword(4)));
+            var expected = reopened.getBukkitView().getTopInventory().getItem(2).clone();
+            transport.forwardToNativeQueue();
+            assertTrue(transport.processor.executeSinglePacket());
+            transport.assertNoErrors(); f.tick();
+            assertTrue(packet.clicks.isEmpty(), "Late revoked exact raw must not fall through to vanilla");
+            assertEquals(expected, reopened.getBukkitView().getTopInventory().getItem(2));
+            assertEquals(sword(4), reopened.getBukkitView().getTopInventory().getItem(0));
+            assertEquals(100, f.nativePlayer.experienceLevel);
+            assertEquals(0, rolls.get());
+            assertTrue(f.player.getItemOnCursor().isEmpty());
+        }
+    }
+
+    @ParameterizedTest @ValueSource(strings = {"replace", "cancel"})
+    void postHandleMutationCannotEraseCapturedForwardAndRetiredWrapperDenial(String mode) throws Exception {
+        try (var f = new NativeAnvilFixture(directory, true);
+                var packet = new NativeAnvilPacketFixture(f);
+                var transport = new NativeAnvilTransportFixture(f, packet)) {
+            f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
+            int before = f.channel.pipeline().names().size();
+            filter(f, event -> post(f, event, forwarded -> {
+                forwarded.setReadOnly(false);
+                if (mode.equals("cancel")) forwarded.setCancelled(true);
+                else forwarded.setPacket(new PacketContainer(forwarded.getPacketType(), inert(f)));
+                f.disableAndCancelTasks();
+            }));
+            transport.receive(raw(f));
+            assertEquals(1, NetworkMarker.getNetworkMarker(transport.received.getFirst()).getPostListeners().size(),
+                    "Only the foreign marker listener exists; ownership never adds an acknowledgement");
+            expire(f);
+            assertEquals(before, f.channel.pipeline().names().size(), "Captured exact raw is still unresolved");
+            when(packet.handler.shouldHandleMessage(any())).thenReturn(false);
+            transport.forwardToNativeQueue();
+            assertEquals(before - 1, f.channel.pipeline().names().size(),
+                    "Late raw is wrapped revoked and releases the retired hook without wrapper.handle");
+            assertTrue(transport.processor.executeSinglePacket());
+            assertFalse(transport.processor.executeSinglePacket());
+            assertTrue(packet.clicks.isEmpty());
+            assertTrue(f.player.getItemOnCursor().isEmpty(), "Skipped revoked wrapper delivers nothing");
+        }
+    }
+
+    @Test void mainFenceSnapshotExcludesSubsequentReceiveWhoseRawIsBehindBarrier() throws Exception {
+        try (var f = new NativeAnvilFixture(directory, true);
+                var packet = new NativeAnvilPacketFixture(f);
+                var transport = new NativeAnvilTransportFixture(f, packet)) {
+            f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
+            var cancellation = filter(f, event -> event.setCancelled(true));
+            transport.receive(raw(f));
+            transport.pumpLibraryTasks(); // First snapshot/barrier precedes the next raw forward.
+            f.protocol.removePacketListener(cancellation);
+            transport.receive(raw(f));
+            assertFalse(transport.received.getLast().isCancelled());
+            transport.forwardToNativeQueue();
+            assertTrue(transport.processor.executeSinglePacket());
+            transport.assertNoErrors();
+            assertEquals(1, packet.clicks.size(), "Later admitted raw must still reach the exact native wrapper");
+            f.tick();
+            assertEquals(6, f.player.getItemOnCursor().getEnchantmentLevel(org.bukkit.enchantments.Enchantment.SHARPNESS));
+        }
+    }
+
+    @Test void unavailablePublicLibrarySchedulerWithholdsEnhancedOutputBeforeHookInstallation() throws Exception {
+        try (var f = new NativeAnvilFixture(directory, true);
+                var packet = new NativeAnvilPacketFixture(f);
+                var transport = new NativeAnvilTransportFixture(f, packet)) {
+            var scheduler = com.comphenix.protocol.ProtocolLibrary.getScheduler();
+            int before = f.channel.pipeline().names().size();
+            try {
+                NativeAnvilFixture.field(com.comphenix.protocol.ProtocolLibrary.class, null, "scheduler", null);
+                f.inputs(sword(5), sword(5));
+                transport.acknowledgeLoop();
+                assertNull(f.output(), "No library-main fence port means no enhanced bypass/output");
+                assertEquals(before, f.channel.pipeline().names().size());
+            } finally {
+                NativeAnvilFixture.field(com.comphenix.protocol.ProtocolLibrary.class, null, "scheduler", scheduler);
+            }
+            f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
+            assertNotNull(f.output(), "A failed constructor did not leave an owned boundary behind");
+        }
+    }
+
+    @Test void rejectedLibraryFenceKeepsUnresolvedTombstoneRatherThanClaimingNoForward() throws Exception {
+        try (var f = new NativeAnvilFixture(directory, true);
+                var packet = new NativeAnvilPacketFixture(f);
+                var transport = new NativeAnvilTransportFixture(f, packet)) {
+            f.inputs(sword(5), sword(5));
+            transport.acknowledgeHook();
+            int before = f.channel.pipeline().names().size();
+            f.librarySchedulingFailure = new IllegalStateException("Controlled library fence rejection");
+            transport.receive(raw(f));
+            assertTrue(transport.received.getFirst().isCancelled());
+            f.disableAndCancelTasks();
+            expire(f);
+            transport.forwardToNativeQueue();
+            assertFalse(transport.processor.executeSinglePacket());
+            assertEquals(before, f.channel.pipeline().names().size(),
+                    "Failed fence cannot prove NO_FORWARD; unresolved ownership stays until raw/disconnect");
+            assertTrue(packet.clicks.isEmpty());
+        }
+    }
+
+    private static void post(NativeAnvilFixture f, PacketEvent event, Consumer<PacketEvent> callback) {
+        // A foreign post listener owns its own marker; production no longer creates one.
+        var marker = NetworkMarker.getNetworkMarker(event);
+        if (marker == null) {
+            marker = new NetworkMarker(ConnectionSide.CLIENT_SIDE, event.getPacketType());
+            event.setNetworkMarker(marker);
+        }
+        marker.addPostListener(new PacketPostListener() {
+            @Override public org.bukkit.plugin.Plugin getPlugin() { return f.plugin; }
+            @Override public void onPostEvent(PacketEvent forwarded) { callback.accept(forwarded); }
+        });
+    }
+
+    private static PacketAdapter filter(NativeAnvilFixture f, Consumer<PacketEvent> callback) {
+        var listener = new PacketAdapter(f.plugin, ListenerPriority.MONITOR, PacketType.Play.Client.WINDOW_CLICK) {
+            @Override public void onPacketReceiving(PacketEvent event) { callback.accept(event); }
+        };
+        f.protocol.addPacketListener(listener);
+        return listener;
+    }
+
+    private static ServerboundContainerClickPacket inert(NativeAnvilFixture f) {
+        return new ServerboundContainerClickPacket(f.menu.containerId, f.menu.getStateId(), (short) -999,
+                (byte) 0, ContainerInput.PICKUP, new it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap<>(), HashedStack.EMPTY);
+    }
+
+    private static void expire(NativeAnvilFixture f) {
+        for (long i = 0; i < AdvancedAnvilProtocolGuard.ADMISSION_TICKS; i++) f.tick();
+    }
+}

src/nativeTest/java/games/dmg/spigotinventoryhelper/NativeAnvilTestChannel.java

@@ -0,0 +1,51 @@
+package games.dmg.spigotinventoryhelper;
+
+import io.netty.channel.*;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/** Real registered local pipeline/event loop, with no byte transport or sockets. */
+final class NativeAnvilTestChannel extends AbstractChannel {
+    final DefaultEventLoop loop = new DefaultEventLoop();
+    private final ChannelConfig config = new DefaultChannelConfig(this);
+    final List<Throwable> failures = new java.util.concurrent.CopyOnWriteArrayList<>();
+    private volatile boolean open = true;
+
+    NativeAnvilTestChannel() throws InterruptedException {
+        super(null);
+        loop.register(this).sync();
+    }
+
+    void acknowledge() throws Exception { loop.submit(() -> {}).get(10, TimeUnit.SECONDS); }
+
+    void finish() throws InterruptedException {
+        close().sync();
+        loop.shutdownGracefully(0, 1, TimeUnit.SECONDS).sync();
+    }
+
+    @Override public ChannelConfig config() { return config; }
+    @Override public ChannelMetadata metadata() { return new ChannelMetadata(false); }
+    @Override public boolean isOpen() { return open; }
+    @Override public boolean isActive() { return open && isRegistered(); }
+    @Override protected boolean isCompatible(EventLoop candidate) { return candidate == loop; }
+    @Override protected java.net.SocketAddress localAddress0() { return null; }
+    @Override protected java.net.SocketAddress remoteAddress0() { return null; }
+    @Override protected void doBind(java.net.SocketAddress address) { throw new UnsupportedOperationException("No sockets"); }
+    @Override protected void doDisconnect() { doClose(); }
+    @Override protected void doClose() { open = false; }
+    @Override protected void doBeginRead() { }
+    @Override protected void doWrite(ChannelOutboundBuffer buffer) { while (buffer.current() != null) buffer.remove(); }
+    @Override protected AbstractUnsafe newUnsafe() {
+        return new AbstractUnsafe() {
+            @Override public void connect(java.net.SocketAddress remote, java.net.SocketAddress local,
+                    ChannelPromise promise) { promise.setFailure(new UnsupportedOperationException("No sockets")); }
+        };
+    }
+    @Override protected DefaultChannelPipeline newChannelPipeline() {
+        return new DefaultChannelPipeline(this) {
+            @Override protected void onUnhandledInboundException(Throwable failure) {
+                if (!(failure instanceof net.minecraft.server.RunningOnDifferentThreadException)) failures.add(failure);
+            }
+        };
+    }
+}

src/nativeTest/java/games/dmg/spigotinventoryhelper/NativeAnvilTransportFixture.java

@@ -8,10 +8,7 @@ import com.comphenix.protocol.events.PacketContainer;
 import com.comphenix.protocol.events.PacketEvent;
 import com.comphenix.protocol.injector.ListenerManager;
 import com.comphenix.protocol.injector.netty.channel.NettyChannelInjector;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.DefaultEventLoop;
-import io.netty.util.DefaultAttributeMap;
+import io.netty.channel.*;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
@@ -19,30 +16,33 @@ import java.util.concurrent.TimeUnit;
 import net.minecraft.network.PacketProcessor;
 import net.minecraft.network.protocol.game.ServerboundContainerClickPacket;
 
-/** Actual ProtocolLib post-adapter forwarding and native thread/queue dispatch, controlled transport ports.
- * Real owned local pipeline, no socket, world/server startup or live-client claim. Strict private library
- * entry is used after packet-type lookup; library callbacks and main->Netty->native-main are real.
+/** Actual library forwarding, one registered real event loop/pipeline, and real native queue.
+ * Only byte I/O, packet lookup/filter invocation and external player/world ports are controlled.
+ * No sockets/server startup. Handler readiness and library-main callbacks require explicit pumping.
  */
 final class NativeAnvilTransportFixture implements AutoCloseable {
     final NativeAnvilPacketFixture packet;
     final PacketProcessor processor;
-    private final DefaultEventLoop loop = new DefaultEventLoop();
+    private final NativeAnvilFixture f;
+    private final DefaultEventLoop loop;
     private final CountDownLatch forward = new CountDownLatch(1);
+    private boolean held;
+    private boolean receiving;
     private final List<Throwable> failures = new java.util.concurrent.CopyOnWriteArrayList<>();
     private final NettyChannelInjector injector;
     private final ChannelHandlerContext context;
     final List<PacketEvent> received = new ArrayList<>();
 
     NativeAnvilTransportFixture(NativeAnvilFixture f, NativeAnvilPacketFixture packet) throws Exception {
+        this.f = f;
         this.packet = packet;
         this.processor = f.processor;
         when(packet.handler.shouldHandleMessage(any())).thenReturn(true);
         doAnswer(inv -> { failures.add(inv.getArgument(1)); return null; }).when(packet.handler).onPacketError(any(), any());
-        var channel = mock(Channel.class);
-        var attributes = new DefaultAttributeMap();
-        when(channel.attr(any())).thenAnswer(inv -> attributes.attr(inv.getArgument(0)));
-        when(channel.eventLoop()).thenReturn(loop);
-        when(channel.closeFuture()).thenReturn(mock(io.netty.channel.ChannelFuture.class));
+        if (!(f.channel instanceof NativeAnvilTestChannel channel)) {
+            throw new AssertionError("Transport must start with its real channel; native handlers are not sharable");
+        }
+        loop = channel.loop;
         var listeners = mock(ListenerManager.class);
         when(listeners.hasMainThreadListener(PacketType.Play.Client.WINDOW_CLICK)).thenReturn(true);
         doAnswer(inv -> {
@@ -54,36 +54,50 @@ final class NativeAnvilTransportFixture implements AutoCloseable {
         injector = new NettyChannelInjector(f.player, packet.connection, channel, listeners,
                 mock(com.comphenix.protocol.injector.netty.channel.InjectionFactory.class),
                 mock(com.comphenix.protocol.error.ErrorReporter.class));
-        f.channel.pipeline().addFirst("protocollib-boundary", new io.netty.channel.ChannelInboundHandlerAdapter());
-        var boundary = f.channel.pipeline().context("protocollib-boundary");
-        context = mock(ChannelHandlerContext.class);
-        when(context.channel()).thenReturn(channel);
-        doAnswer(inv -> {
-            try {
-                boundary.fireChannelRead(inv.getArgument(0));
-                f.channel.checkException();
-            }
-            catch (net.minecraft.server.RunningOnDifferentThreadException expected) {
-                // Native PacketUtils has actually enqueued the packet in our real PacketProcessor.
-            } catch (Throwable failure) { failures.add(failure); }
-            return context;
-        }).when(context).fireChannelRead(any());
-        loop.execute(() -> {
-            try { if (!forward.await(10, TimeUnit.SECONDS)) failures.add(new AssertionError("Forward gate timed out")); }
-            catch (InterruptedException failure) { Thread.currentThread().interrupt(); failures.add(failure); }
-        });
+        channel.pipeline().addFirst("protocollib-boundary", new ChannelInboundHandlerAdapter());
+        acknowledgeLoop();
+        context = channel.pipeline().context("protocollib-boundary");
+        if (context.channel() != packet.connection.channel || context.executor() != channel.eventLoop()
+                || channel.eventLoop() != loop) throw new AssertionError("Library and native loop must be identical");
+    }
+
+    void acknowledgeHook() throws Exception {
+        acknowledgeLoop(); // handlerAdded on the actual event loop, not an inline readiness mock
+        f.tick(); // queued owner readiness/recompute callback on main
+        acknowledgeLoop();
+        if (f.output() == null) throw new AssertionError("Readiness callback did not publish enhanced output");
     }
 
     void receive(ServerboundContainerClickPacket raw) throws Exception {
+        if (!held) {
+            held = true;
+            loop.execute(() -> {
+                try { if (!forward.await(10, TimeUnit.SECONDS)) failures.add(new AssertionError("Forward gate timed out")); }
+                catch (InterruptedException failure) { Thread.currentThread().interrupt(); failures.add(failure); }
+            });
+        }
         var method = NettyChannelInjector.class.getDeclaredMethod("processInboundInternal",
                 ChannelHandlerContext.class, PacketContainer.class);
         method.setAccessible(true);
-        method.invoke(injector, context, new PacketContainer(PacketType.Play.Client.WINDOW_CLICK, raw));
+        receiving = true;
+        try { method.invoke(injector, context, new PacketContainer(PacketType.Play.Client.WINDOW_CLICK, raw)); }
+        finally { receiving = false; }
+    }
+
+    void pumpLibraryTasks() {
+        if (receiving) throw new AssertionError("Fence main callback must run after library callback returns");
+        f.pumpLibraryTasks();
     }
 
     void forwardToNativeQueue() throws Exception {
+        pumpLibraryTasks(); // queues barriers behind captured raw tasks, never inline with filters/posts
         forward.countDown();
+        acknowledgeLoop();
+    }
+
+    void acknowledgeLoop() throws Exception {
         loop.submit(() -> {}).get(10, TimeUnit.SECONDS);
+        failures.addAll(((NativeAnvilTestChannel) f.channel).failures);
         assertNoErrors();
     }
 
@@ -93,11 +107,9 @@ final class NativeAnvilTransportFixture implements AutoCloseable {
 
     @Override public void close() {
         forward.countDown();
-        try { loop.shutdownGracefully(0, 1, TimeUnit.SECONDS).sync(); }
-        catch (InterruptedException failure) {
-            Thread.currentThread().interrupt();
-            throw new AssertionError("Interrupted transport shutdown", failure);
-        }
+        try { ((NativeAnvilTestChannel) f.channel).acknowledge(); }
+        catch (Exception failure) { throw new AssertionError("Transport shutdown acknowledgment failed", failure); }
         assertNoErrors();
     }
+
 }