Merge pull request 'fix(events): bound diagnostic retirement during shutdown' (#12) from fs-sec-10-bound-diagnostic-retirement into main

README.md

@@ -39,6 +39,8 @@ Fantasy Stick independently writes compact CloudEvents 1.0 JSON objects as UTF-8
 
 A stable plugin-owned server UUID and per-lifetime boot/sequence identity distinguish records. Events rotate at UTC midnight, on each writer lifetime and before 128 MiB; an oversized record stays whole in its own segment. Accepted work is first retained in internal pending files and exported at least once with unchanged IDs, so consumers must deduplicate by `(source, id)`. Startup recovers pending work, quarantines malformed/interrupted tails outside `*.jsonl`, and reports incomplete prior lifetimes without inventing a shutdown. Queue overload and pre-durable native-observation crashes remain truthful loss windows; logging never rolls back or replays gameplay.
 
+Feature-outbox aggregate overload diagnostics are bounded, rate-limited and off the gameplay thread. Each outbox gives diagnostics only the existing five-second closing-attempt budget: afterward it stops new aggregate callbacks, interrupts its own cooperative handler and retires maintenance bookkeeping once its intent worker finishes. Late rejected observations remain counted but cannot restart aggregate callbacks or resurrect that closed diagnostic owner; they are never durable/exported. Durable pending intents remain available for replay. A completed closing attempt is **not** proof that aggregate diagnostics were logged or that a supplied handler terminated. An interruption-ignoring trusted in-process handler can still retain its executing thread/callback/plugin classloader (and can stall shared diagnostics while running); Java cannot safely kill it. Pending or interrupted aggregate reports are best-effort, not durable event export acknowledgments. Other blocking storage/sink/error callbacks are not made safely cancellable by this safeguard.
+
 These files have no automatic deletion, compression or retention limit, so disk use grows until operators manage it under a separately approved policy. Current generic records include stable account/character/session IDs and precise locations; later feature-owned event slices may include complete public and recognized private-message text. Restrict event files and backups to operators accordingly. They are not encrypted or cryptographically tamper-evident, and feedback/particle preferences do not disable auditing.
 
 ## Character storage and keepsakes

src/main/java/games/dmg/fantasystick/events/FeatureEventOutbox.java

@@ -57,7 +57,7 @@ final class FeatureEventOutbox implements AutoCloseable {
         synchronized (MAINTENANCE_LIFETIME) {
             OWNERS.remove(owner);
             if (OWNERS.isEmpty()) {
-                REPLAYS.shutdown(); DIAGNOSTICS.shutdown();
+                REPLAYS.shutdown(); DIAGNOSTICS.shutdownNow();
                 REPLAYS = null; DIAGNOSTICS = null;
             }
         }
@@ -84,10 +84,14 @@ final class FeatureEventOutbox implements AutoCloseable {
     private final Path directory;
     private final Function<EventInput, CompletionStage<LocalEventLog.Delivery>> sink;
     private final Consumer<Throwable> failures;
+    private final java.util.concurrent.atomic.AtomicBoolean workerFinished = new java.util.concurrent.atomic.AtomicBoolean();
     private final ExecutorService worker = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS,
         new ArrayBlockingQueue<>(MAX_HANDOFFS), task -> new Thread(task, "fantasy-feature-event-intents"),
         new ThreadPoolExecutor.AbortPolicy()) {
-            @Override protected void terminated() { closeReplayCursor(); }
+            @Override protected void terminated() {
+                try { closeReplayCursor(); }
+                finally { workerFinished.set(true); retire(); }
+            }
         };
     private final CompletableFuture<Void> drained = new CompletableFuture<>();
     private final Set<UUID> inFlight = java.util.concurrent.ConcurrentHashMap.newKeySet();
@@ -100,6 +104,10 @@ final class FeatureEventOutbox implements AutoCloseable {
     private final java.util.concurrent.atomic.AtomicLong rejected = new java.util.concurrent.atomic.AtomicLong();
     private final java.util.concurrent.atomic.AtomicLong deferred = new java.util.concurrent.atomic.AtomicLong();
     private final Object diagnosticFence = new Object();
+    // Terminal after the bounded closing attempt; late rejections cannot resurrect this owner.
+    private volatile boolean diagnosticsStopped;
+    // Guarded by diagnosticFence: interrupt only this owner's active aggregate callback.
+    private Thread diagnosticThread;
     private volatile long lastDiagnostic;
     private volatile boolean replayRequested;
     // The directory cursor is confined to the intent worker; it retains no materialized backlog.
@@ -118,39 +126,66 @@ final class FeatureEventOutbox implements AutoCloseable {
     }
 
     private boolean register() {
+        if (diagnosticsStopped) return false;
         if (!registration.compareAndSet(0, 3)) return true;
         if (!OWNER_SLOTS.tryAcquire()) { registration.set(0); return false; }
-        attach(this); registration.set(1); return true;
+        attach(this); registration.set(1);
+        // Closing may finish while a producer is attaching; retire that raced registration too.
+        if (diagnosticsStopped) retire();
+        return true;
     }
 
     private void retire() {
-        if (!worker.isTerminated() || rejected.get() != 0 || deferred.get() != 0 || diagnosing.get()) return;
+        if (!workerFinished.get() || (!diagnosticsStopped
+                && (rejected.get() != 0 || deferred.get() != 0 || diagnosing.get()))) return;
         if (!registration.compareAndSet(1, 2)) return;
         detach(this); OWNER_SLOTS.release(); registration.set(0);
         // A producer seeing state 2 cannot reattach until removal finishes; its counts remain owned here.
-        if (rejected.get() != 0 || deferred.get() != 0) register();
+        if (!diagnosticsStopped && (rejected.get() != 0 || deferred.get() != 0)) register();
     }
 
     private void rejectObservation() { rejected.incrementAndGet(); register(); }
 
     private void diagnose() {
-        if (!diagnosing.compareAndSet(false, true)) return;
+        synchronized (diagnosticFence) {
+            if (diagnosticsStopped || !diagnosing.compareAndSet(false, true)) return;
+            diagnosticThread = Thread.currentThread();
+        }
         try {
-            long now = System.nanoTime();
-            if (!closingDiagnosis.getAndSet(false) && lastDiagnostic != 0
-                    && now - lastDiagnostic < TimeUnit.SECONDS.toNanos(1)) return;
-            long lost = rejected.getAndSet(0), delayed = deferred.getAndSet(0);
-            if (lost != 0 || delayed != 0) {
-                lastDiagnostic = now;
-                report(new HandoffRejection(lost, delayed));
+            long lost, delayed;
+            synchronized (diagnosticFence) {
+                // Admission/dequeue shares the stop fence: a late count cannot be withdrawn
+                // by a diagnostic pass selected just before terminal shutdown.
+                if (diagnosticsStopped) return;
+                long now = System.nanoTime();
+                if (!closingDiagnosis.getAndSet(false) && lastDiagnostic != 0
+                        && now - lastDiagnostic < TimeUnit.SECONDS.toNanos(1)) return;
+                lost = rejected.getAndSet(0); delayed = deferred.getAndSet(0);
+                if (lost != 0 || delayed != 0) lastDiagnostic = now;
             }
+            if (lost != 0 || delayed != 0) report(new HandoffRejection(lost, delayed));
         } finally {
-            diagnosing.set(false);
-            synchronized (diagnosticFence) { diagnosticFence.notifyAll(); }
+            synchronized (diagnosticFence) {
+                diagnosticThread = null;
+                diagnosing.set(false);
+                // Do not leak our cooperative cancellation into another owner's callback.
+                if (diagnosticsStopped) Thread.interrupted();
+                diagnosticFence.notifyAll();
+            }
             retire();
         }
     }
 
+    private void stopDiagnostics() {
+        synchronized (diagnosticFence) {
+            diagnosticsStopped = true;
+            if (diagnosticThread != null) diagnosticThread.interrupt();
+        }
+        // A non-cooperative handler can still retain its own executing stack/classloader.
+        // Detachment is bookkeeping retirement, never proof that diagnostics were delivered.
+        retire();
+    }
+
     void accepted(EventInput event) {
         if (!admit()) return;
         try { worker.execute(() -> {
@@ -429,15 +464,20 @@ final class FeatureEventOutbox implements AutoCloseable {
                 synchronized (diagnosticFence) {
                     while (rejected.get() != 0 || deferred.get() != 0 || diagnosing.get()) {
                         long remaining = deadline - System.nanoTime();
-                        if (remaining <= 0) break; // Bounded shutdown attempt; maintenance retains the aggregate diagnosis.
+                        if (remaining <= 0) break; // Best-effort diagnostics stop at this lifetime's budget.
                         TimeUnit.NANOSECONDS.timedWait(diagnosticFence, remaining);
                     }
                 }
             } catch (InterruptedException interrupted) {
                 Thread.currentThread().interrupt(); report(interrupted);
-            } finally { retire(); drained.complete(null); }
+            } finally { stopDiagnostics(); drained.complete(null); }
         });
     }
 
+    /** Observations not submitted to an aggregate callback, including late calls after diagnostic retirement.
+     * This is neither a durable-event count nor acknowledgment of earlier callback delivery. */
+    long pendingRejectedObservations() { return rejected.get(); }
+
+    /** Bounded closing attempt, not an acknowledgment of diagnostic delivery or handler termination. */
     CompletionStage<Void> drained() { return drained; }
 }

src/main/java/games/dmg/fantasystick/events/LocalEventLog.java

@@ -60,6 +60,7 @@ public final class LocalEventLog implements AutoCloseable {
     private final java.nio.channels.FileLock writerLock;
     private long sequence;
     private Path segment;
+    private Path damagedSegment;
     private LocalDate segmentDate;
     private long segmentBytes;
     private int segmentNumber;
@@ -203,24 +204,40 @@ public final class LocalEventLog implements AutoCloseable {
         }
         boolean recovered = false;
         for (var path : files) {
-            if (Files.isSymbolicLink(path)) throw new IOException("Symbolic link in Fantasy Stick event segments");
-            var bytes = Files.readAllBytes(path);
-            int prefix=0,cursor=0;while(cursor<bytes.length){int end=cursor;while(end<bytes.length&&bytes[end]!='\n')end++;if(end==bytes.length)break;var line=java.util.Arrays.copyOfRange(bytes,cursor,end+1);if(!validJsonObjectLine(line))break;prefix=end+1;cursor=end+1;}if(prefix==bytes.length)continue;
-            recovered = true;
-            var quarantine = path.resolveSibling(path.getFileName() + ".corrupt-" + UUID.randomUUID());
-            try (var channel = FileChannel.open(quarantine, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)) {
-                var tail = ByteBuffer.wrap(bytes, prefix, bytes.length - prefix);
-                while (tail.hasRemaining()) channel.write(tail);
-                channel.force(true);
-            }
-            try (var channel = FileChannel.open(path, StandardOpenOption.WRITE)) {
-                channel.truncate(prefix);
-                channel.force(true);
-            }
+            recovered |= recoverInterruptedTail(path);
         }
         return recovered;
     }
 
+    /** Quarantine must be forced before truncation; failure leaves pending work and repair state intact. */
+    private boolean recoverInterruptedTail(Path path) throws IOException {
+        if (Files.isSymbolicLink(path)) throw new IOException("Symbolic link in Fantasy Stick event segments");
+        var bytes = Files.readAllBytes(path);
+        int prefix = 0;
+        int cursor = 0;
+        while (cursor < bytes.length) {
+            int end = cursor;
+            while (end < bytes.length && bytes[end] != '\n') end++;
+            if (end == bytes.length) break;
+            var line = java.util.Arrays.copyOfRange(bytes, cursor, end + 1);
+            if (!validJsonObjectLine(line)) break;
+            prefix = end + 1;
+            cursor = end + 1;
+        }
+        if (prefix == bytes.length) return false;
+        var quarantine = path.resolveSibling(path.getFileName() + ".corrupt-" + UUID.randomUUID());
+        try (var channel = FileChannel.open(quarantine, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)) {
+            var tail = ByteBuffer.wrap(bytes, prefix, bytes.length - prefix);
+            while (tail.hasRemaining()) channel.write(tail);
+            channel.force(true);
+        }
+        try (var channel = FileChannel.open(path, StandardOpenOption.WRITE)) {
+            channel.truncate(prefix);
+            channel.force(true);
+        }
+        return true;
+    }
+
     private int recoverPending() throws IOException {
         List<Path> files;
         try (var stream = Files.list(pending)) {
@@ -247,6 +264,10 @@ public final class LocalEventLog implements AutoCloseable {
     }
 
     private void append(byte[] line) throws IOException {
+        if (damagedSegment != null) {
+            recoverInterruptedTail(damagedSegment);
+            damagedSegment = null;
+        }
         var date = LocalDate.ofInstant(clock.instant(), ZoneOffset.UTC);
         if (segment == null || !date.equals(segmentDate)
                 || segmentBytes > 0 && segmentBytes + line.length > maximumSegmentBytes) {
@@ -256,6 +277,14 @@ public final class LocalEventLog implements AutoCloseable {
             var buffer = ByteBuffer.wrap(line);
             while (buffer.hasRemaining()) channel.write(buffer);
             channel.force(true);
+        } catch (IOException failure) {
+            // Even open/force/close failures can leave uncertain bytes. Never reuse this
+            // segment or acknowledge pending work until its tail has been repaired.
+            damagedSegment = segment;
+            segment = null;
+            segmentDate = null;
+            segmentBytes = 0;
+            throw failure;
         }
         segmentBytes += line.length;
     }

src/nativeTest/java/games/dmg/fantasystick/FeatureAuditRetirementNativeTest.java

@@ -0,0 +1,85 @@
+package games.dmg.fantasystick;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static games.dmg.fantasystick.QuestTrackingEventsNativeTest.*;
+import static games.dmg.fantasystick.QuestTrackingSafetyEventsNativeTest.acceptSolo;
+
+import games.dmg.fantasystick.events.LocalEventLog;
+import games.dmg.fantasystick.events.QuestTrackingEventBridge;
+import java.nio.file.Path;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/** Actual packaged disable must cancel the handler wired by the production event bridge. */
+class FeatureAuditRetirementNativeTest {
+    @TempDir Path data;
+    @BeforeAll static void bootstrap() throws Exception { NativeRuntime.bootstrap(); }
+
+    @Test void packagedDisableInterruptsBlockedProductionLoggerWithoutReleasingItFromTheTest() throws Exception {
+        var pluginData = data.resolve("plugin");
+        try (var fixture = new QuestTrackingAuditNativeFixture(pluginData)) {
+            acceptSolo(fixture);
+            var f = fixture.f; var actor = fixture.actor; var quest = fixture.directory.active().getFirst();
+            awaitRows(f, pluginData, "preference.changed", 1);
+            FixtureAuditAcknowledgmentNativeTest.awaitOwnQuiescence(f);
+            var original = (QuestTrackingEventBridge) FantasyStickLifecycleNativeTest.Fixture.field(f.plugin, "questTrackingEvents");
+            var originalOutbox = FantasyStickLifecycleNativeTest.Fixture.field(original, "outbox");
+            var held = new CompletableFuture<LocalEventLog.Delivery>();
+            var full = new CountDownLatch(1024);
+            var blocker = new QuestTrackingEventBridge(input -> { full.countDown(); return held; },
+                data.resolve("held"), original::actor, ignored -> {});
+            var blockerWorker = (ExecutorService) FantasyStickLifecycleNativeTest.Fixture.field(
+                FantasyStickLifecycleNativeTest.Fixture.field(blocker, "outbox"), "worker");
+            var nativeThread = Thread.currentThread();
+            var onCaller = new AtomicBoolean();
+            var entered = new CountDownLatch(1); var interrupted = new CountDownLatch(1);
+            var release = new CountDownLatch(1);
+            var handler = new java.util.logging.Handler() {
+                @Override public void publish(java.util.logging.LogRecord record) {
+                    if (!"Quest tracking event intent failed".equals(record.getMessage())) return;
+                    if (Thread.currentThread() == nativeThread) { onCaller.set(true); return; }
+                    entered.countDown();
+                    try { release.await(); }
+                    catch (InterruptedException expected) { interrupted.countDown(); Thread.currentThread().interrupt(); }
+                }
+                @Override public void flush() {}
+                @Override public void close() {}
+            };
+            try {
+                for (int i = 0; i < 1024; i++) blocker.observed(blocker.request(actor.player, null));
+                assertTrue(full.await(30, TimeUnit.SECONDS));
+                blocker.close(); assertTrue(blockerWorker.awaitTermination(30, TimeUnit.SECONDS));
+                f.plugin.getLogger().addHandler(handler);
+                f.playerCommand.execute(actor.player, "fstick", new String[]{"adventure", "tracking"});
+                assertFalse(onCaller.get()); assertTrue(entered.await(5, TimeUnit.SECONDS));
+                f.playerCommand.execute(actor.player, "fstick", new String[]{"adventure", "tracking", "off"});
+                ReskillEventsNativeTest.await(f, () -> actor.messages.contains("Quest tracking is now off."));
+                assertFalse(fixture.directory.trackingEnabled(quest.id(), actor.core.record.id()));
+                assertTrue(rows(pluginData, "preference.read").isEmpty(), "rejected observation is not exported");
+                held.complete(LocalEventLog.Delivery.REJECTED);
+                f.plugin.setEnabled(false); // Actual packaged cleanup closes the original production outbox.
+                original.drained().toCompletableFuture().get(10, TimeUnit.SECONDS);
+                assertTrue(interrupted.await(1, TimeUnit.SECONDS), "disable must interrupt its own cooperative logger");
+                var ownersField = originalOutbox.getClass().getDeclaredField("OWNERS"); ownersField.setAccessible(true);
+                assertFalse(((Set<?>) ownersField.get(null)).contains(originalOutbox));
+                assertFalse(onCaller.get(), "shutdown cancellation must not move logger I/O onto the native thread");
+                assertTrue(rows(pluginData, "preference.read").isEmpty());
+                try (var files = java.nio.file.Files.list(data.resolve("held/quest-tracking-event-intents"))) {
+                    assertEquals(1024, files.count(), "shutdown must not discard unrelated durable intents");
+                }
+            } finally {
+                release.countDown(); f.plugin.getLogger().removeHandler(handler);
+                held.complete(LocalEventLog.Delivery.REJECTED); blocker.close();
+                assertTrue(blockerWorker.awaitTermination(30, TimeUnit.SECONDS));
+                blocker.drained().toCompletableFuture().get(10, TimeUnit.SECONDS);
+            }
+        }
+    }
+}

src/nativeTest/java/games/dmg/fantasystick/LocalEventPartialWriteNativeTest.java

@@ -0,0 +1,359 @@
+package games.dmg.fantasystick;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import games.dmg.fantasystick.events.EventInput;
+import games.dmg.fantasystick.events.LocalEventLog;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockedStatic;
+
+/** Real file bytes and owned writer callbacks; only the append channel's failure is injected. */
+final class LocalEventPartialWriteNativeTest {
+    @TempDir Path data;
+    @BeforeAll static void bootstrap() throws Exception { NativeRuntime.bootstrap(); }
+
+    @Test void sameLifetimeRetryPreservesPrefixAndExportsExactPendingRecordAfterPartialWrite() throws Exception {
+        var clock = Clock.fixed(Instant.parse("2026-09-26T12:00:00Z"), ZoneOffset.UTC);
+        var diagnostics = new ArrayList<String>();
+        try (var log = LocalEventLog.open(data, clock,
+                new LocalEventLog.RuntimeInfo("Season 4", "26.2", "test"), 8, 134_217_728L, diagnostics::add)) {
+            assertEquals(LocalEventLog.Delivery.EXPORTED, log.ready().toCompletableFuture().get(5, TimeUnit.SECONDS));
+            var original = segments().getFirst();
+            var prefix = Files.readAllBytes(original);
+            var input = event(clock, "original\nUnicode ☃");
+            var pending = data.resolve("events/pending/" + input.id() + ".pending");
+            byte[] exact;
+            try (var fault = new AppendFault(log)) {
+                fault.failNext(1);
+                assertEquals(LocalEventLog.Delivery.DURABLE_PENDING,
+                    log.emit(input).toCompletableFuture().get(5, TimeUnit.SECONDS));
+                exact = Files.readAllBytes(pending);
+                assertEquals(1, fault.failures.get());
+                assertArrayEquals(concat(prefix, Arrays.copyOf(exact, AppendFault.PARTIAL_BYTES)),
+                    Files.readAllBytes(original), "the injected fault must really persist a partial record");
+                assertEquals(LocalEventLog.Delivery.EXPORTED,
+                    log.emit(event(clock, "next")).toCompletableFuture().get(5, TimeUnit.SECONDS));
+            }
+            assertFalse(Files.exists(pending));
+            var rows = exported(); // Strict JSON parser: concatenating the pending record onto its fragment is RED.
+            var recovered = rows.stream().filter(row -> input.id().toString().equals(row.get("id").getAsString())).toList();
+            assertEquals(1, recovered.size());
+            assertEquals(JsonParser.parseString(new String(exact, java.nio.charset.StandardCharsets.UTF_8)), recovered.getFirst());
+            assertTrue(Files.readAllBytes(original).length >= prefix.length);
+            assertArrayEquals(prefix, Arrays.copyOf(Files.readAllBytes(original), prefix.length));
+            assertTrue(corruptEvidence().stream().anyMatch(bytes ->
+                Arrays.equals(Arrays.copyOf(exact, AppendFault.PARTIAL_BYTES), bytes)));
+            assertTrue(diagnostics.stream().anyMatch(message -> message.contains("durable pending")));
+        }
+        exported();
+    }
+
+    @Test void partialWriteDuringPendingReplayRetainsExactProofUntilASecondRecoverySucceeds() throws Exception {
+        var clock = Clock.fixed(Instant.parse("2026-09-26T12:00:00Z"), ZoneOffset.UTC);
+        try (var log = LocalEventLog.open(data, clock,
+                new LocalEventLog.RuntimeInfo("Season 4", "26.2", "test"), 8, 134_217_728L, ignored -> { })) {
+            log.ready().toCompletableFuture().get(5, TimeUnit.SECONDS);
+            var input = event(clock, "pending replay");
+            var pending = data.resolve("events/pending/" + input.id() + ".pending");
+            try (var fault = new AppendFault(log)) {
+                fault.failNext(2);
+                assertEquals(LocalEventLog.Delivery.DURABLE_PENDING,
+                    log.emit(input).toCompletableFuture().get(5, TimeUnit.SECONDS));
+                var exact = Files.readAllBytes(pending);
+                var unaccepted = event(clock, "not yet accepted");
+                assertEquals(LocalEventLog.Delivery.REJECTED,
+                    log.emit(unaccepted).toCompletableFuture().get(5, TimeUnit.SECONDS));
+                assertArrayEquals(exact, Files.readAllBytes(pending));
+                assertEquals(2, fault.failures.get());
+                assertEquals(LocalEventLog.Delivery.EXPORTED,
+                    log.emit(event(clock, "retry")).toCompletableFuture().get(5, TimeUnit.SECONDS));
+                assertFalse(Files.exists(pending));
+                var rows = exported();
+                assertEquals(List.of(JsonParser.parseString(new String(exact, java.nio.charset.StandardCharsets.UTF_8))),
+                    rows.stream().filter(row -> input.id().toString().equals(row.get("id").getAsString())).toList());
+                assertTrue(rows.stream().noneMatch(row -> unaccepted.id().toString().equals(row.get("id").getAsString())));
+                assertEquals(2, corruptEvidence().stream().filter(bytes ->
+                    Arrays.equals(Arrays.copyOf(exact, AppendFault.PARTIAL_BYTES), bytes)).count());
+            }
+        }
+    }
+
+    @Test void failedQuarantineForceCannotTruncateDamageDeletePendingOrAcknowledgeReplay() throws Exception {
+        var clock = Clock.fixed(Instant.parse("2026-09-26T12:00:00Z"), ZoneOffset.UTC);
+        try (var log = LocalEventLog.open(data, clock,
+                new LocalEventLog.RuntimeInfo("Season 4", "26.2", "test"), 8, 134_217_728L, ignored -> { })) {
+            log.ready().toCompletableFuture().get(5, TimeUnit.SECONDS);
+            var original = segments().getFirst();
+            var input = event(clock, "quarantine failure");
+            var pending = data.resolve("events/pending/" + input.id() + ".pending");
+            try (var fault = new AppendFault(log)) {
+                fault.failNext(1);
+                assertEquals(LocalEventLog.Delivery.DURABLE_PENDING,
+                    log.emit(input).toCompletableFuture().get(5, TimeUnit.SECONDS));
+                var exact = Files.readAllBytes(pending);
+                var damaged = Files.readAllBytes(original);
+                fault.quarantineForceFailures.set(1);
+                assertEquals(LocalEventLog.Delivery.REJECTED,
+                    log.emit(event(clock, "repair unavailable")).toCompletableFuture().get(5, TimeUnit.SECONDS));
+                assertEquals(1, fault.failedQuarantines.get());
+                assertArrayEquals(damaged, Files.readAllBytes(original), "unforced quarantine must not authorize truncation");
+                assertArrayEquals(exact, Files.readAllBytes(pending));
+                assertEquals(LocalEventLog.Delivery.EXPORTED,
+                    log.emit(event(clock, "repair available")).toCompletableFuture().get(5, TimeUnit.SECONDS));
+                assertFalse(Files.exists(pending));
+                assertEquals(1, exported().stream().filter(row -> input.id().toString().equals(row.get("id").getAsString())).count());
+                assertTrue(corruptEvidence().stream().anyMatch(bytes ->
+                    Arrays.equals(Arrays.copyOf(exact, AppendFault.PARTIAL_BYTES), bytes)));
+            }
+        }
+    }
+
+    @Test void completeWriteWithFailedForceRetainsPendingAndPermitsOnlyIdenticalAtLeastOnceDuplicates() throws Exception {
+        var clock = Clock.fixed(Instant.parse("2026-09-26T12:00:00Z"), ZoneOffset.UTC);
+        try (var log = LocalEventLog.open(data, clock,
+                new LocalEventLog.RuntimeInfo("Season 4", "26.2", "test"), 8, 134_217_728L, ignored -> { })) {
+            log.ready().toCompletableFuture().get(5, TimeUnit.SECONDS);
+            var input = event(clock, "uncertain force");
+            var pending = data.resolve("events/pending/" + input.id() + ".pending");
+            try (var fault = new AppendFault(log)) {
+                fault.appendForceFailures.set(1);
+                assertEquals(LocalEventLog.Delivery.DURABLE_PENDING,
+                    log.emit(input).toCompletableFuture().get(5, TimeUnit.SECONDS));
+                var exact = Files.readAllBytes(pending);
+                assertEquals(1, fault.failedAppendForces.get());
+                assertEquals(LocalEventLog.Delivery.EXPORTED,
+                    log.emit(event(clock, "retry")).toCompletableFuture().get(5, TimeUnit.SECONDS));
+                var repeated = exported().stream().filter(row -> input.id().toString().equals(row.get("id").getAsString())).toList();
+                assertEquals(2, repeated.size(), "an uncertain completed append may repeat, never promise exactly once");
+                for (var row : repeated) assertEquals(JsonParser.parseString(new String(exact, java.nio.charset.StandardCharsets.UTF_8)), row);
+                assertFalse(Files.exists(pending));
+                assertTrue(corruptEvidence().isEmpty(), "complete valid records must not be quarantined or rewritten");
+            }
+        }
+    }
+
+    @Test void registeredCancelledNativeActionRecoversItsOriginalContextWithoutRestartOrGameplayMutation() throws Exception {
+        try (var fixture = new FantasyStickLifecycleNativeTest.Fixture(data);
+                var nativeFailures = new NativeFailures(fixture)) {
+            fixture.plugin.setEnabled(true);
+            var actor = new FantasyStickLifecycleNativeTest.Actor(data);
+            fixture.players.put(actor.core.id, actor.player);
+            fixture.activate(actor);
+            doReturn(null).when(actor.player).getFishHook(); // No hook exists for this ordinary held-slot event.
+            var log = (LocalEventLog) FantasyStickLifecycleNativeTest.Fixture.field(fixture.plugin, "eventLog");
+            log.ready().toCompletableFuture().get(5, TimeUnit.SECONDS);
+            var before = actor.core.snapshot();
+            var event = new org.bukkit.event.player.PlayerItemHeldEvent(actor.player, 0, 1);
+            event.setCancelled(true);
+            try (var fault = new AppendFault(log)) {
+                fault.onlyContaining = "\"entry_point\":\"PlayerItemHeldEvent\"";
+                fault.failNext(1);
+                fixture.dispatcher.accept(event);
+                fault.worker.submit(() -> { }).get(5, TimeUnit.SECONDS); // Both real registered capture stages settle.
+                assertEquals(1, fault.failures.get());
+                assertTrue(event.isCancelled());
+                assertArrayEquals(before, actor.core.snapshot());
+                var exact = JsonParser.parseString(new String(fault.failedLine, java.nio.charset.StandardCharsets.UTF_8)).getAsJsonObject();
+                assertTrue(exact.get("type").getAsString().endsWith("action.observed"));
+                assertEquals(actor.core.id.toString(), exact.get("playerid").getAsString());
+                assertEquals(actor.core.record.id().toString(), exact.get("characterid").getAsString());
+                var correlated = fault.worker.submit(this::exported).get(5, TimeUnit.SECONDS).stream()
+                    .filter(row -> exact.getAsJsonObject("data").get("action_id")
+                    .equals(row.getAsJsonObject("data").get("action_id"))).toList();
+                assertEquals(2, correlated.size(), "one observation and one resolution, never a duplicated success");
+                assertEquals(List.of(exact), correlated.stream().filter(row -> row.get("id").equals(exact.get("id"))).toList());
+                var resolved = correlated.stream().filter(row -> row.get("type").getAsString().endsWith("action.resolved"))
+                    .findFirst().orElseThrow().getAsJsonObject("data");
+                assertTrue(resolved.get("cancelled").getAsBoolean());
+                assertEquals("cancelled", resolved.get("outcome").getAsString());
+                nativeFailures.assertClean();
+                assertFalse(Files.exists(data.resolve("events/pending/" + exact.get("id").getAsString() + ".pending")));
+                assertTrue(corruptEvidence().stream().anyMatch(bytes ->
+                    Arrays.equals(Arrays.copyOf(fault.failedLine, AppendFault.PARTIAL_BYTES), bytes)));
+            }
+        }
+        assertEquals(1, exported().stream().filter(row -> row.get("type").getAsString().endsWith("game.stopped")).count());
+    }
+
+    private static EventInput event(Clock clock, String marker) {
+        return new EventInput(UUID.randomUUID(), UUID.randomUUID(), clock.instant(), "action.observed", null,
+            Map.of("action_kind", "partial_write_probe", "marker", marker, "outcome", "observed"));
+    }
+
+    private List<Path> segments() throws IOException {
+        try (var files = Files.walk(data.resolve("events"))) {
+            return files.filter(path -> path.getFileName().toString().endsWith(".jsonl")).sorted().toList();
+        }
+    }
+
+    private List<JsonObject> exported() throws IOException {
+        var rows = new ArrayList<JsonObject>();
+        for (var path : segments()) {
+            var bytes = Files.readAllBytes(path);
+            if (bytes.length == 0) continue;
+            assertEquals('\n', bytes[bytes.length - 1], "every exported record must be LF terminated");
+            for (var line : Files.readAllLines(path)) rows.add(JsonParser.parseString(line).getAsJsonObject());
+        }
+        return rows;
+    }
+
+    private List<byte[]> corruptEvidence() throws IOException {
+        var evidence = new ArrayList<byte[]>();
+        try (var files = Files.walk(data.resolve("events"))) {
+            for (var path : files.filter(path -> path.getFileName().toString().contains(".corrupt-")).toList()) {
+                assertFalse(path.getFileName().toString().endsWith(".jsonl"));
+                evidence.add(Files.readAllBytes(path));
+            }
+        }
+        return evidence;
+    }
+
+    private static byte[] concat(byte[] first, byte[] second) {
+        var bytes = Arrays.copyOf(first, first.length + second.length);
+        System.arraycopy(second, 0, bytes, first.length, second.length);
+        return bytes;
+    }
+
+    /** Capture, never suppress, failures from this fixture's native callback boundaries. */
+    private static final class NativeFailures implements AutoCloseable {
+        private final java.util.logging.Logger server;
+        private final java.util.logging.Logger plugin;
+        private final List<java.util.logging.LogRecord> failures = new java.util.concurrent.CopyOnWriteArrayList<>();
+        private final java.util.logging.Handler handler = new java.util.logging.Handler() {
+            @Override public void publish(java.util.logging.LogRecord record) {
+                if (record.getLevel().intValue() >= java.util.logging.Level.SEVERE.intValue()) failures.add(record);
+            }
+            @Override public void flush() { }
+            @Override public void close() { }
+        };
+        private NativeFailures(FantasyStickLifecycleNativeTest.Fixture fixture) {
+            server = fixture.server.getLogger();
+            plugin = fixture.plugin.getLogger();
+            server.addHandler(handler);
+            if (plugin != server) plugin.addHandler(handler);
+        }
+        private void assertClean() {
+            assertTrue(failures.isEmpty(), () -> failures.stream()
+                .map(record -> record.getMessage() + ": " + record.getThrown()).toList().toString());
+        }
+        @Override public void close() {
+            server.removeHandler(handler);
+            if (plugin != server) plugin.removeHandler(handler);
+            assertClean();
+        }
+    }
+
+    /** Scoped to this log's serial worker, never installed globally or on the native/main thread. */
+    private static final class AppendFault implements AutoCloseable {
+        private static final int PARTIAL_BYTES = 47;
+        private final ExecutorService worker;
+        private final MockedStatic<FileChannel> channels;
+        private final AtomicInteger remaining = new AtomicInteger();
+        private final AtomicInteger failures = new AtomicInteger();
+        private final AtomicInteger appendForceFailures = new AtomicInteger();
+        private final AtomicInteger failedAppendForces = new AtomicInteger();
+        private final AtomicInteger quarantineForceFailures = new AtomicInteger();
+        private final AtomicInteger failedQuarantines = new AtomicInteger();
+        private volatile String onlyContaining;
+        private volatile byte[] failedLine;
+
+        private AppendFault(LocalEventLog log) throws Exception {
+            worker = (ExecutorService) FantasyStickLifecycleNativeTest.Fixture.field(log, "worker");
+            channels = worker.submit(() -> {
+                var scoped = mockStatic(FileChannel.class, CALLS_REAL_METHODS);
+                try {
+                    scoped.when(() -> FileChannel.open(any(Path.class), eq(StandardOpenOption.WRITE),
+                        eq(StandardOpenOption.APPEND))).thenAnswer(call -> {
+                            if (remaining.get() == 0 && appendForceFailures.get() == 0) return call.callRealMethod();
+                            Path path = call.getArgument(0);
+                            var real = path.getFileSystem().provider().newFileChannel(path,
+                                Set.of(StandardOpenOption.WRITE, StandardOpenOption.APPEND));
+                            var failing = mock(FileChannel.class);
+                            doAnswer(write -> {
+                                ByteBuffer source = write.getArgument(0);
+                                var complete = new byte[source.remaining()];
+                                source.duplicate().get(complete);
+                                if ((onlyContaining == null || new String(complete, java.nio.charset.StandardCharsets.UTF_8)
+                                        .contains(onlyContaining)) && take(remaining)) {
+                                    failedLine = complete;
+                                    var partial = source.slice();
+                                    partial.limit(Math.min(PARTIAL_BYTES, partial.remaining()));
+                                    while (partial.hasRemaining()) real.write(partial);
+                                    real.force(true);
+                                    failures.incrementAndGet();
+                                    throw new IOException("controlled partial append");
+                                }
+                                return real.write(source);
+                            }).when(failing).write(any(ByteBuffer.class));
+                            doAnswer(force -> {
+                                if (take(appendForceFailures)) {
+                                    failedAppendForces.incrementAndGet();
+                                    throw new IOException("controlled append force failure");
+                                }
+                                real.force(force.getArgument(0));
+                                return null;
+                            }).when(failing).force(anyBoolean());
+                            doAnswer(close -> { real.close(); return null; }).when(failing).close();
+                            return failing;
+                        });
+                    scoped.when(() -> FileChannel.open(any(Path.class), eq(StandardOpenOption.CREATE_NEW),
+                        eq(StandardOpenOption.WRITE))).thenAnswer(call -> {
+                            Path path = call.getArgument(0);
+                            if (!path.getFileName().toString().contains(".corrupt-") || !take(quarantineForceFailures))
+                                return call.callRealMethod();
+                            var real = path.getFileSystem().provider().newFileChannel(path,
+                                Set.of(StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE));
+                            var failing = mock(FileChannel.class);
+                            doAnswer(write -> real.write((ByteBuffer) write.getArgument(0))).when(failing).write(any(ByteBuffer.class));
+                            doAnswer(force -> {
+                                failedQuarantines.incrementAndGet();
+                                throw new IOException("controlled quarantine force failure");
+                            }).when(failing).force(anyBoolean());
+                            doAnswer(close -> { real.close(); return null; }).when(failing).close();
+                            return failing;
+                        });
+                    return scoped;
+                } catch (Throwable failure) {
+                    scoped.close();
+                    throw failure;
+                }
+            }).get(5, TimeUnit.SECONDS);
+        }
+
+        private static boolean take(AtomicInteger count) {
+            return count.getAndUpdate(value -> Math.max(0, value - 1)) > 0;
+        }
+
+        private void failNext(int count) { remaining.set(count); }
+
+        @Override public void close() {
+            try { worker.submit(channels::close).get(5, TimeUnit.SECONDS); }
+            catch (InterruptedException interrupted) { Thread.currentThread().interrupt(); throw new AssertionError(interrupted); }
+            catch (Exception failure) { throw new AssertionError(failure); }
+        }
+    }
+}

src/test/java/games/dmg/fantasystick/events/FeatureEventOutboxCapacityTest.java

@@ -84,11 +84,10 @@ class FeatureEventOutboxCapacityTest {
     @Test void outstandingAcknowledgmentsRemainBoundedAndFailureReleasesCapacityForOriginalIdReplay() throws Exception {
         var held = new CompletableFuture<LocalEventLog.Delivery>();
         var failures = new CopyOnWriteArrayList<Throwable>();
-        var rejectionReports = new CountDownLatch(2);
         var delivered = new CopyOnWriteArrayList<EventInput>();
         var writer = new FeatureEventOutbox(data, "held-intents", Set.of(TYPE), input -> {
             delivered.add(input); return held;
-        }, error -> { failures.add(error); if (error instanceof java.util.concurrent.RejectedExecutionException) rejectionReports.countDown(); });
+        }, failures::add);
         var inputs = new java.util.ArrayList<EventInput>();
         try {
             for (int i = 0; i < CAPACITY + 1; i++) { var input = event(); inputs.add(input); writer.accepted(input); }
@@ -141,9 +140,12 @@ class FeatureEventOutboxCapacityTest {
         assertEquals(CAPACITY, replayed.size());
         assertTrue(replayFailures.isEmpty(), replayFailures.toString());
         try (var files = Files.list(data.resolve("held-intents"))) { assertEquals(List.of(), files.toList()); }
-        writer.accepted(event());
-        assertTrue(rejectionReports.await(5, TimeUnit.SECONDS), "closed handoff diagnosis is also off-thread");
-        assertEquals(2, failures.stream().filter(error -> error instanceof java.util.concurrent.RejectedExecutionException).count(),
-            "closed handoff must remain diagnosed, not silently durable");
+        assertEquals(0, writer.pendingRejectedObservations());
+        var late = event(); writer.accepted(late);
+        assertEquals(1, writer.pendingRejectedObservations(), "late closed rejection remains counted, not falsely logged");
+        assertEquals(1, failures.stream().filter(error -> error instanceof java.util.concurrent.RejectedExecutionException).count(),
+            "terminal diagnostics cannot restart callbacks after their bounded close attempt");
+        assertFalse(Files.exists(data.resolve("held-intents").resolve(late.id() + ".pending")),
+            "closed rejection is never durable/exported");
     }
 }

src/test/java/games/dmg/fantasystick/events/FeatureEventOutboxProgressTest.java

@@ -80,7 +80,8 @@ class FeatureEventOutboxProgressTest {
                 assertTrue(firstPass.await(30, TimeUnit.SECONDS));
                 assertEquals(CAPACITY - 1, seen.size(), "held callbacks and the replay command share the bound");
                 acknowledge.complete(null);
-                assertTrue(exported.await(30, TimeUnit.SECONDS), "returned capacity must continue durable replay automatically");
+                assertTrue(exported.await(30, TimeUnit.SECONDS), () -> "returned capacity must continue durable replay automatically; submitted="
+                    + seen.size() + "/" + original.size() + "; remaining exports=" + exported.getCount() + "; diagnostics=" + failures);
             } finally { acknowledge.complete(null); closeOwned(reader); }
             assertEquals(original.size(), seen.size());
             assertEquals(Set.copyOf(original), Set.copyOf(seen));

src/test/java/games/dmg/fantasystick/events/FeatureEventOutboxRetirementTest.java

@@ -0,0 +1,219 @@
+package games.dmg.fantasystick.events;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/** Real saturation, callback and shutdown boundaries, including their actual resource lifetime. */
+class FeatureEventOutboxRetirementTest {
+    @TempDir Path data;
+    private static final String TYPE = "action.observed";
+
+    private static EventInput event() {
+        return new EventInput(UUID.randomUUID(), UUID.randomUUID(), Instant.EPOCH, TYPE, null,
+            Map.of("outcome", "observed"));
+    }
+
+    private static Object field(Object owner, String name) throws Exception {
+        var field = FeatureEventOutbox.class.getDeclaredField(name);
+        field.setAccessible(true);
+        return field.get(owner);
+    }
+
+    private static ExecutorService maintenance(String name) throws Exception {
+        return (ExecutorService) field(null, name);
+    }
+
+    private static boolean registered(FeatureEventOutbox owner) throws Exception {
+        return ((Set<?>) field(null, "OWNERS")).contains(owner);
+    }
+
+    private static void closeOwned(FeatureEventOutbox owner) throws Exception {
+        owner.close();
+        assertTrue(((ExecutorService) field(owner, "worker")).awaitTermination(30, TimeUnit.SECONDS));
+        owner.drained().toCompletableFuture().get(8, TimeUnit.SECONDS);
+    }
+
+    @Test void concurrentLateCallsRemainCountedWithoutRevivingCallbacksOrMaintenance() throws Exception {
+        var callbacks = new java.util.concurrent.atomic.AtomicInteger();
+        var owner = new FeatureEventOutbox(data, "late-owner", Set.of(TYPE), input -> {
+            callbacks.incrementAndGet(); return CompletableFuture.completedFuture(LocalEventLog.Delivery.EXPORTED);
+        }, ignored -> callbacks.incrementAndGet());
+        var replays = maintenance("REPLAYS"); var diagnostics = maintenance("DIAGNOSTICS");
+        closeOwned(owner);
+        assertTrue(replays.isShutdown()); assertTrue(diagnostics.isShutdown());
+        var producers = new CompletableFuture<?>[4];
+        for (int i = 0; i < producers.length; i++) producers[i] = CompletableFuture.runAsync(() -> {
+            for (int observation = 0; observation < 2500; observation++) owner.accepted(event());
+        });
+        CompletableFuture.allOf(producers).get(5, TimeUnit.SECONDS);
+        assertEquals(10000, owner.pendingRejectedObservations());
+        assertEquals(0, callbacks.get(), "no sink or diagnostic call may start in a retired lifetime");
+        assertFalse(registered(owner));
+        assertNull(maintenance("REPLAYS")); assertNull(maintenance("DIAGNOSTICS"));
+        assertFalse(Files.exists(data.resolve("late-owner")), "late observations cannot create durable intent files");
+    }
+
+    @Test void nonCooperativeHandlerDoesNotRetainRegistryAndLateCallbacksCannotCloseFreshLifetime() throws Exception {
+        var held = new CompletableFuture<LocalEventLog.Delivery>();
+        var blockerFull = new CountDownLatch(1022);
+        var full = new CountDownLatch(2);
+        var entered = new CountDownLatch(1);
+        var interruptSeen = new CountDownLatch(1);
+        var returned = new CountDownLatch(1);
+        var release = new CountDownLatch(1);
+        var saved = new java.util.concurrent.CopyOnWriteArrayList<EventInput>();
+        var attempted = new java.util.concurrent.atomic.AtomicInteger();
+        var unexpected = new AtomicReference<Throwable>();
+        var owner = new FeatureEventOutbox(data, "non-cooperative", Set.of(TYPE), input -> {
+            saved.add(input); full.countDown(); return held;
+        }, failure -> {
+            if (!(failure instanceof FeatureEventOutbox.HandoffRejection)) { unexpected.set(failure); return; }
+            attempted.incrementAndGet(); entered.countDown();
+            boolean released = false;
+            while (!released) {
+                try { release.await(); released = true; }
+                catch (InterruptedException ignored) { interruptSeen.countDown(); }
+            }
+            returned.countDown();
+        });
+        var blocker = new FeatureEventOutbox(data, "other-durable", Set.of(TYPE), input -> {
+            blockerFull.countDown(); return held;
+        }, unexpected::set);
+        var replays = maintenance("REPLAYS");
+        var diagnostics = maintenance("DIAGNOSTICS");
+        FeatureEventOutbox fresh = null;
+        try {
+            for (int i = 0; i < 1022; i++) blocker.accepted(event());
+            assertTrue(blockerFull.await(30, TimeUnit.SECONDS));
+            closeOwned(blocker); // Unsettled acknowledgments still occupy capacity, without a live maintenance owner.
+            owner.accepted(event()); owner.accepted(event());
+            assertTrue(full.await(30, TimeUnit.SECONDS));
+            owner.accepted(event()); assertTrue(entered.await(5, TimeUnit.SECONDS));
+            for (int i = 0; i < 5000; i++) owner.accepted(event());
+            assertEquals(1, attempted.get(), "stalled diagnostics cannot queue per-observation callbacks");
+            held.complete(LocalEventLog.Delivery.REJECTED);
+            closeOwned(owner);
+            assertFalse(registered(owner));
+            assertTrue(interruptSeen.await(1, TimeUnit.SECONDS));
+            assertEquals(1, returned.getCount(), "drained is NOT acknowledgment of handler completion/log delivery");
+            assertTrue(replays.isShutdown()); assertTrue(diagnostics.isShutdown());
+            assertFalse(diagnostics.awaitTermination(100, TimeUnit.MILLISECONDS),
+                "an interruption-ignoring handler still retains its executing thread: explicit residual limit");
+            try (var files = Files.list(data.resolve("non-cooperative"))) { assertEquals(2, files.count()); }
+            try (var files = Files.list(data.resolve("other-durable"))) { assertEquals(1022, files.count()); }
+            var exported = new CountDownLatch(saved.size());
+            try (var log = LocalEventLog.open(data.resolve("recovery-stream"), java.time.Clock.systemUTC(),
+                    new LocalEventLog.RuntimeInfo("retirement", "native", "test"), 1024, 134_217_728L, ignored -> {})) {
+                assertEquals(LocalEventLog.Delivery.EXPORTED, log.ready().toCompletableFuture().get(10, TimeUnit.SECONDS));
+                fresh = new FeatureEventOutbox(data, "non-cooperative", Set.of(TYPE), input ->
+                    log.emit(input).whenComplete((delivery, failure) -> {
+                        if (failure == null && delivery == LocalEventLog.Delivery.EXPORTED) exported.countDown();
+                    }), unexpected::set);
+                var freshReplays = maintenance("REPLAYS");
+                var freshDiagnostics = maintenance("DIAGNOSTICS");
+                assertNotSame(replays, freshReplays); assertNotSame(diagnostics, freshDiagnostics);
+                for (int i = 0; i < 5000; i++) owner.accepted(event()); // Late native observations cannot reattach the retired owner.
+                owner.close(); assertFalse(registered(owner));
+                assertEquals(10000, owner.pendingRejectedObservations(), "both pre-retirement pending and late rejections remain counted");
+                fresh.replay();
+                assertTrue(exported.await(30, TimeUnit.SECONDS), () -> "remaining exports=" + exported.getCount() + "; failure=" + unexpected.get());
+                release.countDown(); assertTrue(returned.await(5, TimeUnit.SECONDS));
+                assertTrue(diagnostics.awaitTermination(5, TimeUnit.SECONDS));
+                assertFalse(freshReplays.isShutdown()); assertFalse(freshDiagnostics.isShutdown());
+                assertTrue(registered(fresh), "old callback completion cannot retire another lifetime");
+                closeOwned(fresh);
+            }
+            var rows = new java.util.ArrayList<com.google.gson.JsonObject>();
+            try (var files = Files.walk(data.resolve("recovery-stream/events"))) {
+                for (var file : files.filter(path -> path.getFileName().toString().endsWith(".jsonl")).toList())
+                    for (var line : Files.readAllLines(file)) {
+                        var row = com.google.gson.JsonParser.parseString(line).getAsJsonObject();
+                        if ((LocalEventLog.TYPE_PREFIX + TYPE).equals(row.get("type").getAsString())) rows.add(row);
+                    }
+            }
+            assertEquals(saved.size(), rows.size());
+            var expected = saved.stream().collect(java.util.stream.Collectors.toMap(input -> input.id().toString(), input -> input));
+            assertEquals(expected.keySet(), rows.stream().map(row -> row.get("id").getAsString()).collect(java.util.stream.Collectors.toSet()));
+            for (var row : rows) {
+                var original = expected.get(row.get("id").getAsString());
+                assertEquals(original.time().toString(), row.get("time").getAsString());
+                assertEquals(original.data().get("outcome"), row.getAsJsonObject("data").get("outcome").getAsString());
+            }
+            try (var files = Files.list(data.resolve("non-cooperative"))) { assertEquals(0, files.count()); }
+            assertEquals(1, attempted.get(), "abandoned diagnostics are not claimed flushed by another lifetime");
+            assertNull(unexpected.get());
+        } finally {
+            release.countDown(); held.complete(LocalEventLog.Delivery.REJECTED);
+            closeOwned(owner); closeOwned(blocker); if (fresh != null) closeOwned(fresh);
+        }
+    }
+
+    @Test void closedOwnerInterruptsItsCooperativeHandlerWithoutClosingAnotherOwnersMaintenance() throws Exception {
+        var held = new CompletableFuture<LocalEventLog.Delivery>();
+        var full = new CountDownLatch(1024);
+        var entered = new CountDownLatch(1);
+        var interrupted = new CountDownLatch(1);
+        var release = new CountDownLatch(1);
+        var unexpected = new AtomicReference<Throwable>();
+        var owner = new FeatureEventOutbox(data, "closed-owner", Set.of(TYPE), input -> {
+            full.countDown(); return held;
+        }, failure -> {
+            if (!(failure instanceof FeatureEventOutbox.HandoffRejection)) { unexpected.set(failure); return; }
+            entered.countDown();
+            try { release.await(); }
+            catch (InterruptedException expected) { interrupted.countDown(); Thread.currentThread().interrupt(); }
+        });
+        var otherDelivered = new CountDownLatch(1);
+        var otherReported = new CountDownLatch(1);
+        var otherInterrupted = new java.util.concurrent.atomic.AtomicBoolean();
+        var other = new FeatureEventOutbox(data, "live-owner", Set.of(TYPE), input -> {
+            otherDelivered.countDown(); return CompletableFuture.completedFuture(LocalEventLog.Delivery.EXPORTED);
+        }, failure -> {
+            if (failure instanceof FeatureEventOutbox.HandoffRejection) {
+                otherInterrupted.set(Thread.currentThread().isInterrupted()); otherReported.countDown();
+            } else unexpected.set(failure);
+        });
+        var replays = maintenance("REPLAYS");
+        var diagnostics = maintenance("DIAGNOSTICS");
+        try {
+            for (int i = 0; i < 1024; i++) owner.accepted(event());
+            assertTrue(full.await(30, TimeUnit.SECONDS));
+            owner.accepted(event());
+            assertTrue(entered.await(5, TimeUnit.SECONDS));
+            other.accepted(event()); // Its aggregate waits behind the blocked owner, not on the producer.
+            held.complete(LocalEventLog.Delivery.REJECTED);
+            closeOwned(owner); // Do NOT release the handler: bounded close must interrupt it itself.
+            assertTrue(interrupted.await(1, TimeUnit.SECONDS), "cooperative handler must receive shutdown interruption");
+            assertFalse(registered(owner), "closed owner must leave maintenance even with a stalled handler");
+            assertTrue(registered(other));
+            assertFalse(replays.isShutdown());
+            assertFalse(diagnostics.isShutdown());
+            assertTrue(otherReported.await(5, TimeUnit.SECONDS));
+            assertFalse(otherInterrupted.get(), "our cancellation must not leak into an unrelated handler");
+            other.accepted(event());
+            assertTrue(otherDelivered.await(5, TimeUnit.SECONDS));
+            try (var files = Files.list(data.resolve("closed-owner"))) {
+                assertEquals(1024, files.count(), "rejected sink delivery must retain all durable intents");
+            }
+            assertNull(unexpected.get());
+        } finally {
+            release.countDown(); held.complete(LocalEventLog.Delivery.REJECTED);
+            closeOwned(owner); closeOwned(other);
+        }
+        assertTrue(replays.awaitTermination(5, TimeUnit.SECONDS));
+        assertTrue(diagnostics.awaitTermination(5, TimeUnit.SECONDS));
+    }
+}