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/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/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));
+    }
+}